""" ══════════════════════════════════════════════════════════════════════════════ 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 ''}")