Spaces:
Running
Running
| """ | |
| 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() | |