from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import joblib import json import numpy as np import pandas as pd from pathlib import Path app = FastAPI( title="ClimaShock API", description="Pakistan's first distributed causal climate-economic intelligence system", version="1.0.0" ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # load models on startup BASE = Path(__file__).parent / "models" xgb_inf = joblib.load(BASE / "xgb_inf.pkl") xgb_crop = joblib.load(BASE / "xgb_crop.pkl") feat_scaler = joblib.load(BASE / "feat_scaler.pkl") target_scaler = joblib.load(BASE / "target_scaler.pkl") district_metrics = joblib.load(BASE / "district_metrics.pkl") with open(BASE / "ensemble_config.json") as f: ensemble_config = json.load(f) causal_links = pd.read_csv(BASE / "causal_links.csv") climate_df = pd.read_csv(BASE / "pakistan_climate.csv") # district-specific climate norms district_stats = climate_df.groupby("district").agg( rain_mean = ("PRECTOTCORR", "mean"), rain_std = ("PRECTOTCORR", "std"), temp_mean = ("T2M", "mean"), temp_std = ("T2M", "std"), ).to_dict("index") # historical reference values for XGBoost features climate_annual = climate_df.groupby("year").agg( rain_mean = ("PRECTOTCORR", "mean"), rain_std = ("PRECTOTCORR", "std"), temp_mean = ("T2M", "mean"), temp_max = ("T2M_MAX", "mean"), humidity = ("RH2M", "mean"), ).reset_index() HIST_RAIN_MEAN = float(climate_annual["rain_mean"].mean()) HIST_RAIN_STD = float(climate_annual["rain_mean"].std()) HIST_TEMP_MEAN = float(climate_annual["temp_mean"].mean()) HIST_TEMP_STD = float(climate_annual["temp_mean"].std()) # granger coefficients from analysis GRANGER_RAIN_INF = 0.347 GRANGER_TEMP_INF = 0.128 INF_MEAN = 12.8 INF_STD = 7.4 class PredictRequest(BaseModel): district: str rainfall_mm_day: float temperature_c: float gdp_growth_pct: float = 3.5 agri_gdp_pct: float = 22.0 class PredictResponse(BaseModel): district: str rain_zscore: float temp_zscore: float risk_level: str risk_score: float immediate_crop_pct: float inflation_predicted: float inflation_range_low: float inflation_range_high: float gdp_outlook: str cascade_chain: list model_confidence: str @app.get("/") def root(): return { "system": "ClimaShock", "version": "1.0.0", "status": "online", "endpoints": ["/predict", "/causal", "/districts", "/health"] } @app.get("/health") def health(): return {"status": "ok", "models_loaded": True} @app.post("/predict", response_model=PredictResponse) def predict(req: PredictRequest): district = req.district.strip().title() if district not in district_stats: raise HTTPException( status_code=400, detail=f"District '{district}' not found. Available: {list(district_stats.keys())}" ) stats = district_stats[district] # district-specific z-scores rain_z = (req.rainfall_mm_day - stats["rain_mean"]) / max(stats["rain_std"], 0.01) temp_z = (req.temperature_c - stats["temp_mean"]) / max(stats["temp_std"], 0.01) # risk classification risk_score = abs(rain_z) * 0.7 + abs(temp_z) * 0.3 if abs(rain_z) > 2.5: risk_level = "EXTREME" elif abs(rain_z) > 1.5: risk_level = "HIGH" elif abs(rain_z) > 0.5: risk_level = "MODERATE" else: risk_level = "NORMAL" # granger-calibrated inflation prediction inflation_pred = INF_MEAN inflation_pred += GRANGER_RAIN_INF * rain_z * INF_STD inflation_pred += GRANGER_TEMP_INF * temp_z * INF_STD inflation_pred = max(0, round(inflation_pred, 1)) # confidence interval — ±1 std scaled by risk margin = INF_STD * 0.5 * (1 + abs(rain_z) * 0.1) inf_low = round(max(0, inflation_pred - margin), 1) inf_high = round(inflation_pred + margin, 1) # immediate crop impact immediate_crop = round(-abs(rain_z) * 5.2 if rain_z > 1.5 else rain_z * 1.8, 1) # gdp outlook if inflation_pred > 20: gdp_outlook = "contraction likely" elif inflation_pred > 12: gdp_outlook = "slowdown possible" else: gdp_outlook = "stable" # cascade chain — from granger discoveries cascade = [] if abs(rain_z) > 0.5: cascade.append({ "step": 1, "event": "Climate Anomaly Detected", "detail": f"Rainfall {rain_z:+.2f}σ from district norm", "timeframe": "Now", "severity": risk_level, }) if abs(rain_z) > 1.0: cascade.append({ "step": 2, "event": "Crop Stress", "detail": f"Expected yield change: {immediate_crop:+.1f}%", "timeframe": "4–8 weeks", "severity": "HIGH" if immediate_crop < -10 else "MODERATE", }) cascade.append({ "step": 3, "event": "Inflation Pressure", "detail": f"Predicted: {inflation_pred}% (range {inf_low}–{inf_high}%)", "timeframe": "6–18 months (lag=2y Granger)", "severity": "HIGH" if inflation_pred > 15 else "MODERATE", }) cascade.append({ "step": 4, "event": "GDP Outlook", "detail": gdp_outlook.title(), "timeframe": "12–24 months", "severity": "HIGH" if gdp_outlook == "contraction likely" else "LOW", }) confidence = ( "HIGH — well within training distribution" if abs(rain_z) < 2 else "MEDIUM — near extreme historical values" if abs(rain_z) < 3 else "LOW — beyond historical training range" ) return PredictResponse( district = district, rain_zscore = round(rain_z, 3), temp_zscore = round(temp_z, 3), risk_level = risk_level, risk_score = round(risk_score, 3), immediate_crop_pct = immediate_crop, inflation_predicted = inflation_pred, inflation_range_low = inf_low, inflation_range_high= inf_high, gdp_outlook = gdp_outlook, cascade_chain = cascade, model_confidence = confidence, ) @app.get("/causal") def get_causal_links(min_strength: float = 0.08): links = causal_links[causal_links["strength"] >= min_strength].copy() return { "total_links": len(links), "links": links.sort_values("strength", ascending=False).to_dict("records"), "top_discovery": { "cause": "rain_anomaly", "effect": "inflation_pct", "lag_years": 2, "strength": 0.347, "meaning": "Extreme rainfall causes inflation spike after 2-year lag in Pakistan" } } @app.get("/districts") def get_districts(): out = [] for district, stats in district_stats.items(): met = district_metrics.get(district, {}) out.append({ "district": district, "rain_mean": round(stats["rain_mean"], 3), "rain_std": round(stats["rain_std"], 3), "temp_mean": round(stats["temp_mean"], 3), "model_inf_mae": met.get("inf_mae", None), "model_crop_mae": met.get("crop_mae", None), }) return {"districts": out, "count": len(out)} @app.get("/discoveries") def get_discoveries(): return { "discoveries": [ { "rank": 1, "title": "Flood → Inflation Lag", "finding": "Extreme rainfall causes national inflation spike after 2-year lag", "evidence": "Granger causality strength 0.347 — strongest link in system", "case": "2022 Sindh floods → 2023 Pakistan inflation 30.8%", "novel": True, }, { "rank": 2, "title": "Sukkur Epicenter", "finding": "Sukkur is Pakistan highest climate-economic risk zone", "evidence": "2022 rainfall +1293% — highest z-score of all 10 districts", "case": "Cotton -37.7%, Rice -21.5% in single flood year", "novel": True, }, { "rank": 3, "title": "GDP Contraction Predicted", "finding": "LSTM correctly predicted 2023 GDP contraction direction", "evidence": "Inflation MAE 2.87% — predicted 26% vs actual 30.8%", "case": "Predicted 'contraction likely' → actual GDP -0.41%", "novel": False, }, ] }