agent2_property_risk.py

"""
══════════════════════════════════════════════════════════════════════════════
AGENT 2 — Property Risk Assessment Agent  (Step 2 of 5)
══════════════════════════════════════════════════════════════════════════════
PURPOSE  : Geo-spatial peril scoring + structural risk assessment.
           Reads KYC decision from Agent 1 Silver output.
           If this agent DECLINES, Steps 3-5 are logged as SKIPPED.

INPUT    : silver/kyc_decisions/{sub_id}_kyc.json
OUTPUT   : silver/property_risk/{sub_id}_property.json

DECISION : RISK_ACCEPTABLE  → pipeline continues to Agent 3
           RISK_DECLINED    → pipeline halts, submission DECLINED

TRAINING : XGBoost multi-output regressor → individual peril scores (fire,
           flood, wind, liability).  Threshold classifier on top for accept/decline.

PERIL THRESHOLDS (carrier appetite):
    Wind  > 70  → DECLINE
    Flood > 75  → DECLINE
    Fire  > 74  → DECLINE
    Overall > 80 → DECLINE
══════════════════════════════════════════════════════════════════════════════
"""

import json
import pickle
import datetime
import numpy as np
import pandas as pd
# mysql.connector kept as fallback; primary driver is PyMySQL via SQLAlchemy
import mysql.connector
try:
    from sqlalchemy import create_engine, text
    from urllib.parse import quote_plus as _qp
    SQLALCHEMY_AVAILABLE = True
except ImportError:
    SQLALCHEMY_AVAILABLE = False
from pathlib import Path
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_absolute_error, r2_score, classification_report
from sklearn.preprocessing import LabelEncoder
import xgboost as xgb

# ─── CONFIG ──────────────────────────────────────────────────────────────────
# ─── DB CONFIG — supports local MySQL and HuggingFace + Clever Cloud ─────────
import os as _os

def _is_huggingface() -> bool:
    return (
        _os.environ.get("SPACE_ID")            is not None
        or _os.environ.get("HUGGINGFACE_SPACE") is not None
        or _os.environ.get("MYSQL_ADDON_HOST")  is not None
        or _os.environ.get("MYSQL_HOST")        is not None
    )

def _env(addon_key: str, generic_key: str, default: str = "") -> str:
    """Reads MYSQL_ADDON_* first (Clever Cloud), then MYSQL_* (generic), then default."""
    return _os.environ.get(addon_key) or _os.environ.get(generic_key) or default

if _is_huggingface():
    DB = dict(
        host     = _env("MYSQL_ADDON_HOST",     "MYSQL_HOST"),
        port     = int(_env("MYSQL_ADDON_PORT", "MYSQL_PORT", "3306")),
        user     = _env("MYSQL_ADDON_USER",     "MYSQL_USER"),
        password = _env("MYSQL_ADDON_PASSWORD", "MYSQL_PASSWORD"),
        database = _env("MYSQL_ADDON_DB",       "MYSQL_DATABASE"),
    )
else:
    DB = dict(host="localhost", port=3306, user="root", password="root@123", database="bronze")

def T(layer: str, table: str) -> str:
    """
    Returns the correct table reference for the active environment.
    HuggingFace (single schema):  `bronze_submissions`
    Local (separate schemas):     `bronze`.`submissions`
    """
    return f"`{layer}_{table}`" if _is_huggingface() else f"`{layer}`.`{table}`"
MODEL_PATH  = Path("models/agent2_property_risk.pkl")
SILVER_IN   = Path("silver/kyc_decisions")
SILVER_OUT  = Path("silver/property_risk")
MODEL_PATH.parent.mkdir(exist_ok=True)
SILVER_OUT.mkdir(parents=True, exist_ok=True)

# Carrier appetite thresholds
THRESHOLDS = {"wind": 70, "flood": 75, "fire": 74, "overall": 80}

# ── Geo-peril base scores per state (derived from FEMA + historical cat data) ──
STATE_PERILS = {
    # state: (wind_base, flood_base, fire_base, hail_base, quake_base)
    "FL": (80, 72, 15, 20, 5),   "TX": (65, 60, 40, 55, 10),
    "LA": (75, 78, 20, 25, 5),   "NC": (55, 50, 25, 30, 8),
    "SC": (58, 52, 28, 28, 6),   "GA": (45, 42, 30, 35, 10),
    "AL": (52, 48, 28, 38, 8),   "MS": (60, 65, 22, 32, 5),
    "CA": (20, 25, 78, 15, 65),  "AZ": (15, 10, 60, 20, 8),
    "CO": (35, 20, 55, 45, 12),  "WA": (30, 35, 50, 15, 40),
    "IL": (40, 38, 20, 48, 10),  "NY": (35, 40, 18, 22, 8),
    "PA": (30, 35, 15, 25, 6),   "KS": (50, 40, 30, 60, 8),
    "NV": (10, 8,  55, 12, 20),  "OH": (32, 30, 15, 40, 8),
}
DEFAULT_PERIL = (35, 35, 30, 30, 10)

# Construction risk multipliers
CONSTRUCTION_RISK = {
    "Frame":             1.30,
    "Masonry":           0.80,
    "Masonry Veneer":    0.90,
    "Non-Combustible":   0.75,
    "Fire Resistive":    0.65,
    "Log/Timber":        1.40,
    "Prefabricated":     1.20,
}
ROOF_RISK = {
    "Asphalt Shingles": 1.10,
    "Metal":            0.80,
    "Tile":             0.85,
    "Slate":            0.75,
    "Wood Shake":       1.35,
    "Flat/Built-Up":    1.20,
    "EPDM Membrane":    1.05,
}
PROPERTY_TYPE_RISK = {
    "Single Family":       1.00,
    "Condo Unit":          0.85,
    "Townhouse":           0.92,
    "Mobile Home":         1.45,
    "Vacation / Seasonal": 1.25,
    "Multi-Family":        1.10,
    "Commercial Building": 1.15,
}

# ─── FEATURE ENGINEERING ─────────────────────────────────────────────────────
def extract_property_features(df: pd.DataFrame) -> pd.DataFrame:
    """
    Derive property risk features from Bronze property + submission columns.
    """
    feats = pd.DataFrame()
    state = df.get("state_code", df.get("state", pd.Series(["XX"]*len(df))))

    # ── Peril base scores from state lookup ──
    def get_peril(s, idx, default):
        return STATE_PERILS.get(str(s).upper(), DEFAULT_PERIL)[idx]

    feats["wind_base"]    = state.apply(lambda s: get_peril(s, 0, 35))
    feats["flood_base"]   = state.apply(lambda s: get_peril(s, 1, 35))
    feats["fire_base"]    = state.apply(lambda s: get_peril(s, 2, 30))
    feats["hail_base"]    = state.apply(lambda s: get_peril(s, 3, 30))
    feats["quake_base"]   = state.apply(lambda s: get_peril(s, 4, 10))

    # ── Construction risk modifier (encoded as multiplier × 100) ──
    const = df.get("construction_type", pd.Series(["Frame"]*len(df)))
    feats["construction_risk"] = (const.map(CONSTRUCTION_RISK).fillna(1.0) * 100).astype(int)

    # ── Roof type risk ──
    roof = df.get("roof_type", pd.Series(["Asphalt Shingles"]*len(df)))
    feats["roof_risk"] = (roof.map(ROOF_RISK).fillna(1.0) * 100).astype(int)

    # ── Property type risk ──
    ptype = df.get("property_type", pd.Series(["Single Family"]*len(df)))
    feats["property_type_risk"] = (ptype.map(PROPERTY_TYPE_RISK).fillna(1.0) * 100).astype(int)

    # ── Property age (older = higher risk) ──
    year_built = pd.to_numeric(df.get("year_built", pd.Series([1990]*len(df))), errors="coerce").fillna(1990)
    feats["property_age_years"] = (2024 - year_built).clip(0, 150)

    # ── Roof age ──
    roof_year  = pd.to_numeric(df.get("roof_year", pd.Series([2010]*len(df))), errors="coerce").fillna(2010)
    feats["roof_age_years"] = (2024 - roof_year).clip(0, 50)

    # ── Square footage (normalised) ──
    feats["sqft_norm"] = (
        pd.to_numeric(df.get("square_footage", pd.Series([1800]*len(df))), errors="coerce")
        .fillna(1800).clip(500, 10000) / 1000
    )

    # ── Number of stories ──
    feats["num_stories"] = pd.to_numeric(
        df.get("num_stories", pd.Series([1]*len(df))), errors="coerce").fillna(1).clip(1, 10)

    # ── Coverage limit (proxy for property value / exposure) ──
    feats["coverage_limit_k"] = (
        pd.to_numeric(df.get("requested_coverage_limit", pd.Series([300_000]*len(df))), errors="coerce")
        .fillna(300_000) / 1_000
    )

    # ── Coastal state flag ──
    coastal = {"FL", "TX", "LA", "NC", "SC", "GA", "AL", "MS", "CA", "NY", "WA"}
    feats["is_coastal"] = state.apply(lambda s: 1 if str(s).upper() in coastal else 0)

    # ── High wildfire state flag ──
    fire_states = {"CA", "AZ", "CO", "WA", "NV", "OR", "MT", "ID"}
    feats["is_high_fire"] = state.apply(lambda s: 1 if str(s).upper() in fire_states else 0)

    # ── High tornado state flag ──
    tornado = {"TX", "KS", "OK", "NE", "IA", "MO", "IL", "AR"}
    feats["is_tornado_belt"] = state.apply(lambda s: 1 if str(s).upper() in tornado else 0)

    # ── Older property with wood shake roof (compound risk) ──
    roof_col = df.get("roof_type", pd.Series(["Asphalt Shingles"]*len(df))).fillna("Asphalt Shingles")
    feats["old_wood_shake"] = (
        (feats["property_age_years"] > 40) & (roof_col == "Wood Shake")
    ).astype(int)

    return feats

# ─── DERIVE TARGET SCORES FROM BRONZE ────────────────────────────────────────
def derive_property_targets(df: pd.DataFrame) -> pd.DataFrame:
    """
    Reconstruct peril scores from Bronze data. In production these come from
    FEMA API / third-party geo services; here we simulate them from the
    decision outcomes encoded in raw_payload.
    """
    targets = pd.DataFrame()

    def parse_prop_payload(row):
        try:
            p    = json.loads(row["raw_payload"])
            prop = p.get("property", {})
            return {
                "prop_risk_score":   prop.get("prop_risk_score"),
                "risk_band":         prop.get("risk_band"),
                "final_outcome":     p.get("agent_results", {}).get("final_outcome"),
            }
        except Exception:
            return {"prop_risk_score": None, "risk_band": None, "final_outcome": None}

    parsed   = df.apply(parse_prop_payload, axis=1, result_type="expand")
    state    = df.get("state_code", pd.Series(["XX"]*len(df)))

    # Re-derive individual peril scores using base + noise
    # (In production these come from geo APIs — here we reconstruct from overall score)
    np.random.seed(42)
    n = len(df)
    overall = pd.to_numeric(parsed["prop_risk_score"], errors="coerce").fillna(35)

    # Distribute overall score into perils based on state risk profile
    targets["wind_score"]   = (overall * state.map({k: v[0]/100 for k, v in STATE_PERILS.items()}).fillna(0.45)
                                + np.random.uniform(-5, 5, n)).clip(0, 100).round(1)
    targets["flood_score"]  = (overall * state.map({k: v[1]/100 for k, v in STATE_PERILS.items()}).fillna(0.40)
                                + np.random.uniform(-5, 5, n)).clip(0, 100).round(1)
    targets["fire_score"]   = (overall * state.map({k: v[2]/100 for k, v in STATE_PERILS.items()}).fillna(0.35)
                                + np.random.uniform(-4, 4, n)).clip(0, 100).round(1)
    targets["overall_risk"] = overall.round(1)

    # Binary: 1 = RISK_ACCEPTABLE, 0 = RISK_DECLINED
    # Use pipeline_status = 'PROP_DECLINED' as the ground truth label
    # (this is the exact value written by the Bronze data generator)
    is_prop_decline = (
        df.get("pipeline_status", pd.Series(["ISSUED"]*n)) == "PROP_DECLINED"
    )
    targets["risk_label"] = (~is_prop_decline).astype(int)  # 1=acceptable, 0=declined
    print(f"  Risk ACCEPTABLE: {targets['risk_label'].sum()} | DECLINED: {(targets['risk_label']==0).sum()}")
    return targets

# ─── DATA LOADING ────────────────────────────────────────────────────────────
def load_bronze_property_data() -> pd.DataFrame:
    print("Connecting to Bronze MySQL...")
    if SQLALCHEMY_AVAILABLE:
        _pwd = _qp(DB['password'])
        eng  = create_engine(
            f"mysql+pymysql://{DB['user']}:{_pwd}@{DB['host']}:{DB['port']}/{DB['database']}?charset=utf8mb4",
            pool_pre_ping=True, pool_recycle=280
        )
        conn = eng.connect()
    else:
        conn = mysql.connector.connect(**DB)
    query = f"""
        
        SELECT
            s.submission_id,
            s.pipeline_status,
            s.halt_reason,
            s.requested_coverage_limit,
            s.final_outcome,
            s.raw_payload,
            pr.state_code,
            pr.property_type,
            pr.construction_type,
            pr.roof_type,
            pr.roof_year,
            pr.year_built,
            pr.square_footage,
            pr.num_stories,
            pr.occupancy
        FROM {T('bronze','submissions')}  s
        JOIN {T('bronze','properties')}   pr ON s.property_id = pr.property_id
        WHERE s.submitted_at BETWEEN '2024-01-01' AND '2024-12-31 23:59:59'
        ORDER BY s.submitted_at
    """
    df = pd.read_sql(query, conn)
    conn.close()
    print(f"  Loaded {len(df)} Bronze records")
    return df

# ─── TRAINING ────────────────────────────────────────────────────────────────
def train_property_model():
    print("\n" + "═"*60)
    print("AGENT 2 — Property Risk Model Training")
    print("═"*60)

    df      = load_bronze_property_data()
    X       = extract_property_features(df)
    targets = derive_property_targets(df)

    FEATURES = X.columns.tolist()
    print(f"\nFeatures ({len(FEATURES)}): {FEATURES}")

    models    = {}
    metrics   = {}

    # Train one regressor per peril score + one binary classifier
    for target in ["wind_score", "flood_score", "fire_score", "overall_risk"]:
        y = targets[target]
        X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.2, random_state=42)
        m = xgb.XGBRegressor(n_estimators=200, max_depth=4, learning_rate=0.05,
                               subsample=0.8, colsample_bytree=0.8,
                               eval_metric="rmse", early_stopping_rounds=15,
                               random_state=42, verbosity=0)
        m.fit(X_tr, y_tr, eval_set=[(X_te, y_te)], verbose=False)
        pred = m.predict(X_te)
        mae  = mean_absolute_error(y_te, pred)
        r2   = r2_score(y_te, pred)
        models[target]  = m
        metrics[target] = {"MAE": round(mae, 2), "R2": round(r2, 4)}
        print(f"  {target:<18}  MAE={mae:.2f}  R²={r2:.4f}")

    # Binary classifier: risk_acceptable
    y_cls = targets["risk_label"]
    X_tr, X_te, y_tr, y_te = train_test_split(X, y_cls, test_size=0.2, stratify=y_cls, random_state=42)
    cls = xgb.XGBClassifier(n_estimators=250, max_depth=4, learning_rate=0.05,
                              subsample=0.8, colsample_bytree=0.8,
                              eval_metric="auc", early_stopping_rounds=15,
                              random_state=42, verbosity=0)
    cls.fit(X_tr, y_tr, eval_set=[(X_te, y_te)], verbose=False)
    y_pred = cls.predict(X_te)
    print("\n  Risk Classifier Report:")
    print(classification_report(y_te, y_pred, target_names=["RISK_DECLINED", "RISK_ACCEPTABLE"]))

    artefact = {
        "models":     models,
        "classifier": cls,
        "features":   FEATURES,
        "thresholds": THRESHOLDS,
        "trained_at": datetime.datetime.now().isoformat(),
        "version":    "1.0",
    }
    with open(MODEL_PATH, "wb") as f:
        pickle.dump(artefact, f)
    print(f"\n  Model saved → {MODEL_PATH}")
    return artefact

# ─── INFERENCE ───────────────────────────────────────────────────────────────
def run_property_risk_agent(kyc_decision: dict, submission_json: dict) -> dict:
    """
    Parameters
    ----------
    kyc_decision    : dict  Output from Agent 1 (silver/kyc_decisions/)
    submission_json : dict  Full Bronze JSON payload

    Returns
    -------
    dict  Property risk decision written to silver/property_risk/
    """
    sub_id = submission_json.get("submission_id", "UNKNOWN")

    # Guard: should not run if KYC failed
    if kyc_decision.get("status") != "KYC_PASS":
        return {"submission_id": sub_id, "status": "SKIPPED",
                "skip_reason": "KYC_FAIL — pipeline halted at Step 1"}

    prop    = submission_json.get("property", {})
    policy  = submission_json.get("policy_request", {})

    row = pd.DataFrame([{
        "state_code":                prop.get("state", "XX"),
        "construction_type":         prop.get("construction_type", "Frame"),
        "roof_type":                 prop.get("roof_type", "Asphalt Shingles"),
        "property_type":             prop.get("property_type", "Single Family"),
        "year_built":                prop.get("year_built", 1990),
        "roof_year":                 prop.get("roof_year", 2010),
        "square_footage":            prop.get("square_footage", 1800),
        "num_stories":               prop.get("num_stories", 1),
        "requested_coverage_limit":  policy.get("limit", 300_000),
    }])

    feats = extract_property_features(row)

    try:
        with open(MODEL_PATH, "rb") as f:
            art = pickle.load(f)

        feat_cols = art["features"]
        f_in      = feats[feat_cols]

        wind_score    = float(art["models"]["wind_score"].predict(f_in)[0])
        flood_score   = float(art["models"]["flood_score"].predict(f_in)[0])
        fire_score    = float(art["models"]["fire_score"].predict(f_in)[0])
        overall_risk  = float(art["models"]["overall_risk"].predict(f_in)[0])
        risk_prob     = float(art["classifier"].predict_proba(f_in)[0, 1])  # P(acceptable)

    except FileNotFoundError:
        # Model not yet trained — use rule-based scores only
        state = prop.get("state", "XX").upper()
        bases = STATE_PERILS.get(state, DEFAULT_PERIL)
        wind_score   = float(bases[0]) + np.random.uniform(-5, 5)
        flood_score  = float(bases[1]) + np.random.uniform(-5, 5)
        fire_score   = float(bases[2]) + np.random.uniform(-4, 4)
        overall_risk = (wind_score + flood_score + fire_score) / 3
        risk_prob    = 0.90 if overall_risk < 60 else 0.45

    # Round
    wind_score   = round(max(0, min(100, wind_score)), 1)
    flood_score  = round(max(0, min(100, flood_score)), 1)
    fire_score   = round(max(0, min(100, fire_score)), 1)
    overall_risk = round(max(0, min(100, overall_risk)), 1)

    # ── Determine risk band ──
    if overall_risk <= 35:    risk_band = "LOW"
    elif overall_risk <= 60:  risk_band = "MEDIUM"
    elif overall_risk <= 80:  risk_band = "HIGH"
    else:                     risk_band = "DECLINED"

    # ── Hard threshold decline rules ──
    decline_reason = None
    if wind_score  > THRESHOLDS["wind"]:
        decline_reason = f"WIND-001 — Wind risk score {wind_score:.0f}/100 exceeds carrier maximum {THRESHOLDS['wind']}"
    elif flood_score > THRESHOLDS["flood"]:
        decline_reason = f"FLOOD-002 — Flood risk score {flood_score:.0f}/100 exceeds carrier maximum {THRESHOLDS['flood']}"
    elif fire_score  > THRESHOLDS["fire"]:
        decline_reason = f"FIRE-003 — Wildfire risk score {fire_score:.0f}/100 exceeds carrier maximum {THRESHOLDS['fire']}"
    elif overall_risk > THRESHOLDS["overall"]:
        decline_reason = f"OVERALL-004 — Combined risk score {overall_risk:.0f}/100 exceeds carrier maximum {THRESHOLDS['overall']}"

    status = "RISK_DECLINED" if decline_reason else "RISK_ACCEPTABLE"

    output = {
        "submission_id":  sub_id,
        "agent":          "Property_Risk_Agent",
        "step":           2,
        "status":         status,
        "decision":       "DECLINE" if decline_reason else "ACCEPT",
        "decline_reason": decline_reason,
        "peril_scores": {
            "wind_score":   wind_score,
            "flood_score":  flood_score,
            "fire_score":   fire_score,
            "overall_risk": overall_risk,
        },
        "risk_band":           risk_band,
        "risk_acceptability_prob": round(risk_prob, 4),
        "thresholds_applied":  THRESHOLDS,
        "processed_at":        datetime.datetime.now().isoformat(),
        "next_step":           "Underwriting_Agent" if not decline_reason else "PIPELINE_HALTED",
        "s3_output_uri":       f"s3://pcins-silver/property_risk/{sub_id}_property.json",
    }

    out_file = SILVER_OUT / f"{sub_id}_property.json"
    with open(out_file, "w") as f:
        json.dump(output, f, indent=2)
    return output

# ─── MAIN ────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    train_property_model()

    print("\n" + "─"*60)
    print("SMOKE TESTS")
    print("─"*60)

    kyc_pass = {"status": "KYC_PASS", "credit_score": 720}

    tests = [
        {   # Should ACCEPT — low risk inland state
            "submission_id": "SUB-TEST-001",
            "property":      {"state": "PA", "construction_type": "Masonry", "roof_type": "Tile",
                              "property_type": "Single Family", "year_built": 2010,
                              "roof_year": 2015, "square_footage": 2000, "num_stories": 2},
            "policy_request": {"limit": 350_000},
        },
        {   # Should DECLINE — coastal FL high wind
            "submission_id": "SUB-TEST-002",
            "property":      {"state": "FL", "construction_type": "Frame", "roof_type": "Wood Shake",
                              "property_type": "Vacation / Seasonal", "year_built": 1970,
                              "roof_year": 2000, "square_footage": 1200, "num_stories": 1},
            "policy_request": {"limit": 800_000},
        },
        {   # Should DECLINE — high wildfire CA
            "submission_id": "SUB-TEST-003",
            "property":      {"state": "CA", "construction_type": "Log/Timber", "roof_type": "Wood Shake",
                              "property_type": "Single Family", "year_built": 1965,
                              "roof_year": 1998, "square_footage": 2500, "num_stories": 1},
            "policy_request": {"limit": 1_200_000},
        },
    ]

    for t in tests:
        r = run_property_risk_agent(kyc_pass, t)
        print(f"  {t['submission_id']} | {t['property']['state']} "
              f"| Wind:{r['peril_scores']['wind_score']:.0f} "
              f"Flood:{r['peril_scores']['flood_score']:.0f} "
              f"Fire:{r['peril_scores']['fire_score']:.0f} "
              f"→ {r['status']}  {r['decline_reason'] or ''}")