""" GridGuard AI — Live Demo App (Gradio) ====================================== Run locally: python app.py Deploy on HF Spaces: push this file + data/ + models/ to a Space (SDK: Gradio). requirements.txt is provided alongside this file. Tabs: 1. Grid Overview — transformer loss dashboard (Level 1 detection) 2. Flagged Households — risk-ranked table (Levels 2+3 combined) 3. Investigate a Household — type a household_id, see its full risk profile 4. Try a Synthetic Household — sliders to simulate a new household and see its live fraud-risk score (good for an audience "what if" demo) """ import os import joblib import numpy as np import pandas as pd import gradio as gr import plotly.express as px BASE = os.path.dirname(__file__) DATA_DIR = os.path.join(BASE, "data") MODEL_DIR = os.path.join(BASE, "models") households_risk = pd.read_csv(os.path.join(DATA_DIR, "household_risk_scores.csv")) tx_monthly = pd.read_csv(os.path.join(DATA_DIR, "transformers.csv")) readings = pd.read_csv(os.path.join(DATA_DIR, "readings_monthly.csv")) iso = joblib.load(os.path.join(MODEL_DIR, "isolation_forest.joblib")) scaler = joblib.load(os.path.join(MODEL_DIR, "scaler.joblib")) feature_cols = joblib.load(os.path.join(MODEL_DIR, "feature_cols.joblib")) LATEST_MONTH_IDX = tx_monthly["month_idx"].max() # ---------------------------------------------------------------------------- # Tab 1 — Grid overview # ---------------------------------------------------------------------------- def grid_overview(): latest = tx_monthly[tx_monthly.month_idx == LATEST_MONTH_IDX].copy() latest["loss_pct_display"] = (latest["loss_pct"] * 100).round(1) latest = latest.sort_values("loss_pct", ascending=False) fig = px.bar( latest, x="transformer_id", y="loss_pct_display", title="Transformer Loss % — Latest Month (>8% suggests theft beyond normal technical loss)", labels={"loss_pct_display": "Loss %", "transformer_id": "Transformer"}, color="loss_pct_display", color_continuous_scale="Reds", ) fig.add_hline(y=8, line_dash="dash", line_color="black", annotation_text="Normal technical loss ceiling (8%)") table = latest[["transformer_id", "energy_injected_kwh", "sum_metered_kwh", "unaccounted_kwh", "loss_pct_display"]] table.columns = ["Transformer", "Energy Injected (kWh)", "Metered Total (kWh)", "Unaccounted (kWh)", "Loss %"] return fig, table.round(1) # ---------------------------------------------------------------------------- # Tab 2 — Flagged households table # ---------------------------------------------------------------------------- def flagged_households(min_tier): tier_order = {"Low": 0, "Medium": 1, "High": 2} df = households_risk.copy() df = df[df["risk_tier"].map(tier_order) >= tier_order[min_tier]] df = df.sort_values("fraud_risk_score", ascending=False) cols = ["household_id", "transformer_id", "neighbourhood_id", "income_class", "avg_last3_kwh", "drop_ratio", "peer_zscore", "transformer_loss_pct", "fraud_risk_score", "risk_tier"] out = df[cols].copy() out["drop_ratio"] = (out["drop_ratio"] * 100).round(1) out["transformer_loss_pct"] = (out["transformer_loss_pct"] * 100).round(1) out["fraud_risk_score"] = out["fraud_risk_score"].round(1) out["peer_zscore"] = out["peer_zscore"].round(2) out.columns = ["Household", "Transformer", "Neighbourhood", "Income Class", "Avg Last-3mo (kWh)", "Drop vs Prior (%)", "Peer Z-score", "Transformer Loss (%)", "Fraud Risk Score", "Risk Tier"] return out # ---------------------------------------------------------------------------- # Tab 3 — Investigate a single household by ID # ---------------------------------------------------------------------------- def investigate_household(household_id): if household_id not in households_risk["household_id"].values: return None, "Household ID not found. Try one like HH00001.", None row = households_risk[households_risk.household_id == household_id].iloc[0] hist = readings[readings.household_id == household_id].sort_values("month_idx") fig = px.line( hist, x="month", y="metered_kwh", markers=True, title=f"{household_id} — Metered Consumption Over Time" ) fig.update_yaxes(title="kWh") summary = ( f"**Transformer:** {row.transformer_id} | **Neighbourhood:** {row.neighbourhood_id} | " f"**Income class:** {row.income_class}\n\n" f"**Fraud Risk Score:** {row.fraud_risk_score:.1f}/100 → **Tier: {row.risk_tier}**\n\n" f"- Drop vs prior period: **{row.drop_ratio*100:.1f}%**\n" f"- Peer z-score (vs similar neighbours): **{row.peer_zscore:.2f}**\n" f"- Transformer currently flagged for excess loss: **{'Yes' if row.transformer_flagged else 'No'}** " f"({row.transformer_loss_pct*100:.1f}% loss)\n" f"- Unsupervised anomaly score: **{row.anomaly_score:.2f}** (0=normal, 1=highly anomalous)\n\n" f"_Ground-truth label (synthetic data only): " f"{'FRAUD INJECTED' if row.is_fraud_ground_truth else 'genuine'}_" ) return fig, summary, None # ---------------------------------------------------------------------------- # Tab 4 — Simulate a new household with sliders # ---------------------------------------------------------------------------- def simulate_household(prior_avg_kwh, last3_avg_kwh, peer_mean_kwh, peer_std_kwh, transformer_loss_pct): drop_ratio = max(0.0, (prior_avg_kwh - last3_avg_kwh) / prior_avg_kwh) if prior_avg_kwh > 0 else 0.0 peer_std_kwh = peer_std_kwh if peer_std_kwh > 0 else 1.0 peer_zscore = (last3_avg_kwh - peer_mean_kwh) / peer_std_kwh trend_slope = (last3_avg_kwh - prior_avg_kwh) / 9.0 # approx over 9-month gap x = np.array([[drop_ratio, peer_zscore, trend_slope, transformer_loss_pct / 100]]) x_scaled = scaler.transform(x) raw = -iso.decision_function(x_scaled)[0] # rescale roughly using training distribution stats baked into the model's typical range anomaly_score = 1 / (1 + np.exp(-3 * raw)) # logistic squashing for a stable 0-1 demo display norm_drop = np.clip(drop_ratio, 0, 1) norm_peer = np.clip(-peer_zscore / 3, 0, 1) tx_boost = (0.15 if transformer_loss_pct > 8 else 0.0) risk_score = float(np.clip(100 * (0.45 * anomaly_score + 0.30 * norm_drop + 0.25 * norm_peer + tx_boost), 0, 100)) tier = "High" if risk_score > 70 else ("Medium" if risk_score > 40 else "Low") explanation = ( f"### Fraud Risk Score: {risk_score:.1f}/100 → **{tier} risk**\n\n" f"- Consumption drop vs prior period: **{drop_ratio*100:.1f}%**\n" f"- Peer z-score: **{peer_zscore:.2f}** " f"({'well below neighbours' if peer_zscore < -1 else 'within normal peer range'})\n" f"- Transformer loss context: **{transformer_loss_pct:.1f}%** " f"({'above normal — supports investigation' if transformer_loss_pct > 8 else 'normal range'})\n" ) return explanation # ---------------------------------------------------------------------------- # Build interface # ---------------------------------------------------------------------------- with gr.Blocks(title="GridGuard AI — Energy Theft Detection") as demo: gr.Markdown( "# ⚡ GridGuard AI\n" "**AI-powered revenue assurance & energy-theft detection for distribution utilities**\n\n" "Proof of concept built on fully synthetic data, demonstrating a hierarchical detection " "approach: transformer-level energy balance → neighbourhood peer comparison → household " "behavioural anomaly detection, combined into a single Fraud Risk Score for field-agent triage." ) with gr.Tab("1. Grid Overview"): gr.Markdown("Sum of metered readings vs. energy injected into each transformer reveals " "feeders losing more than the expected technical loss range.") overview_btn = gr.Button("Load Grid Overview") overview_plot = gr.Plot() overview_table = gr.Dataframe() overview_btn.click(grid_overview, outputs=[overview_plot, overview_table]) with gr.Tab("2. Flagged Households"): gr.Markdown("Households ranked by combined Fraud Risk Score. Field agents should " "prioritise **High** tier; **Medium** tier suits a lighter remote audit.") tier_filter = gr.Radio(["Low", "Medium", "High"], value="Medium", label="Show households at or above tier") flagged_table = gr.Dataframe() tier_filter.change(flagged_households, inputs=tier_filter, outputs=flagged_table) demo.load(flagged_households, inputs=tier_filter, outputs=flagged_table) with gr.Tab("3. Investigate a Household"): hh_input = gr.Textbox(label="Household ID", placeholder="e.g. HH00001", value=households_risk[households_risk.risk_tier == "High"].iloc[0].household_id) hh_btn = gr.Button("Investigate") hh_plot = gr.Plot() hh_summary = gr.Markdown() hh_dummy = gr.Textbox(visible=False) hh_btn.click(investigate_household, inputs=hh_input, outputs=[hh_plot, hh_summary, hh_dummy]) with gr.Tab("4. Simulate a Household"): gr.Markdown("Move the sliders to simulate a household's consumption profile and " "see the model's live risk assessment — good for an audience Q&A demo.") with gr.Row(): with gr.Column(): prior_avg = gr.Slider(20, 800, value=300, label="Avg consumption — prior period (kWh/month)") last3_avg = gr.Slider(0, 800, value=280, label="Avg consumption — last 3 months (kWh/month)") peer_mean = gr.Slider(20, 800, value=300, label="Neighbourhood peer average (kWh/month)") peer_std = gr.Slider(5, 200, value=60, label="Neighbourhood peer std-dev (kWh)") tx_loss = gr.Slider(0, 30, value=6, label="Transformer loss % (this month)") with gr.Column(): sim_output = gr.Markdown() sim_btn = gr.Button("Compute Risk Score", variant="primary") sim_btn.click(simulate_household, inputs=[prior_avg, last3_avg, peer_mean, peer_std, tx_loss], outputs=sim_output) gr.Markdown( "---\n" "*Synthetic proof-of-concept — Session 2: 'Circuit Breaker: From Student to Builder', " "NIEEES Sensitization Seminar, University of Jos.*" ) if __name__ == "__main__": demo.launch()