""" ClaimSense — Agent 10: Severity & Reserve Agent ================================================= XGBoost + GLM ensemble for claim severity classification and initial reserve estimation. Training: call train_severity_model() Inference: call run_severity_agent(fnol_result, coverage_result, fraud_result, submission) """ import os, logging import numpy as np import pandas as pd log = logging.getLogger(__name__) MODELS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'models') MODEL_PATH = os.path.join(MODELS_DIR, 'agent10_severity.pkl') SEVERITY_BANDS = ['MINOR', 'MODERATE', 'MAJOR', 'CATASTROPHIC'] FEATURE_COLS = [ 'estimated_damage_norm', 'coverage_limit_norm', 'amount_ratio', 'fraud_score', 'days_to_report', 'incident_type_encoded', 'property_age', 'deductible_norm', 'has_police_report', 'credit_score_norm', 'net_payable_norm', ] def _encode_incident(t: str) -> int: return {'FIRE':4,'EARTHQUAKE':4,'FLOOD':4,'STRUCTURAL':3, 'WIND':3,'HAIL':2,'WATER':2,'THEFT':2, 'LIABILITY':1,'VANDALISM':1,'OTHER':1}.get(str(t).upper(), 1) def _build_features(fnol: dict, coverage: dict, fraud: dict, submission: dict) -> dict: norm = fnol.get('normalised_fields', {}) estimated_dmg = float(norm.get('estimated_damage') or 0) coverage_limit= float(coverage.get('applicable_limit') or 300000) deductible = float(coverage.get('deductible') or 2500) net_payable = float(coverage.get('net_payable_est') or 0) fraud_score = int(fraud.get('fraud_score') or 0) days_to_report= int(fnol.get('days_to_report') or 0) prop = (submission.get('property') or {}) yr_built = int(prop.get('year_built') or 2000) prop_age = max(0, 2026 - yr_built) credit_score = float((submission.get('insured') or {}).get('credit_score') or 650) has_police = int(bool((norm.get('has_police_report')))) return { 'estimated_damage_norm': min(estimated_dmg / 500000, 3.0), 'coverage_limit_norm': coverage_limit / 1000000, 'amount_ratio': min(estimated_dmg / max(coverage_limit, 1), 3.0), 'fraud_score': fraud_score, 'days_to_report': days_to_report, 'incident_type_encoded': _encode_incident(fnol.get('incident_type', 'OTHER')), 'property_age': min(prop_age, 100), 'deductible_norm': deductible / 50000, 'has_police_report': has_police, 'credit_score_norm': credit_score / 850, 'net_payable_norm': min(net_payable / 500000, 3.0), } def _generate_synthetic_training(n=2500, seed=7): rng = np.random.default_rng(seed) rows = [] for _ in range(n): coverage_limit = float(rng.choice([150000,250000,350000,500000,750000,1000000])) deductible = float(rng.choice([1000,2500,5000,10000])) incident_enc = int(rng.integers(1, 5)) prop_age = int(rng.integers(0, 80)) fraud_sc = int(rng.integers(0, 100)) days_rep = int(rng.integers(0, 120)) credit_sc = int(rng.integers(450, 820)) # Base damage driven by incident severity base = {4: 0.65, 3: 0.35, 2: 0.18, 1: 0.08}.get(incident_enc, 0.1) estimated_dmg = float(coverage_limit * rng.uniform(base * 0.5, base * 1.5)) estimated_dmg = min(estimated_dmg, coverage_limit * 1.1) net_payable = max(0, min(estimated_dmg, coverage_limit) - deductible) # Severity band ground truth pct = estimated_dmg / coverage_limit if pct < 0.10: band = 'MINOR' elif pct < 0.35: band = 'MODERATE' elif pct < 0.70: band = 'MAJOR' else: band = 'CATASTROPHIC' # Final payout (slightly below estimate after adjustment) adj_factor = rng.uniform(0.70, 0.95) final_payout = round(net_payable * adj_factor, 2) rows.append({ 'estimated_damage_norm': min(estimated_dmg / 500000, 3.0), 'coverage_limit_norm': coverage_limit / 1000000, 'amount_ratio': min(estimated_dmg / coverage_limit, 3.0), 'fraud_score': fraud_sc, 'days_to_report': days_rep, 'incident_type_encoded': incident_enc, 'property_age': prop_age, 'deductible_norm': deductible / 50000, 'has_police_report': int(rng.random() < 0.6), 'credit_score_norm': credit_sc / 850, 'net_payable_norm': min(net_payable / 500000, 3.0), 'severity_band': band, 'final_payout': final_payout, }) return pd.DataFrame(rows) def train_severity_model(): import joblib from xgboost import XGBClassifier, XGBRegressor from sklearn.model_selection import train_test_split from sklearn.preprocessing import LabelEncoder from sklearn.metrics import accuracy_score, mean_absolute_error os.makedirs(MODELS_DIR, exist_ok=True) log.info("[SEVERITY] Generating synthetic training data...") df = _generate_synthetic_training(n=3000) le = LabelEncoder() le.fit(SEVERITY_BANDS) df['severity_encoded'] = le.transform(df['severity_band']) X = df[FEATURE_COLS] yc = df['severity_encoded'] yr = df['final_payout'] X_tr, X_te, yc_tr, yc_te, yr_tr, yr_te = train_test_split( X, yc, yr, test_size=0.2, random_state=42 ) clf = XGBClassifier( n_estimators=200, max_depth=5, learning_rate=0.07, subsample=0.85, colsample_bytree=0.85, eval_metric='mlogloss', random_state=42, verbosity=0 ) clf.fit(X_tr, yc_tr) acc = accuracy_score(yc_te, clf.predict(X_te)) log.info(f"[SEVERITY] Classifier accuracy: {acc:.3f}") reg = XGBRegressor( n_estimators=200, max_depth=5, learning_rate=0.07, subsample=0.85, random_state=42, verbosity=0 ) reg.fit(X_tr, yr_tr) mae = mean_absolute_error(yr_te, reg.predict(X_te)) log.info(f"[SEVERITY] Reserve regressor MAE: ${mae:,.0f}") model = { 'classifier': clf, 'regressor': reg, 'label_encoder': le, 'feature_cols': FEATURE_COLS, 'accuracy': acc, 'mae': mae } joblib.dump(model, MODEL_PATH) log.info(f"[SEVERITY] Saved → {MODEL_PATH}") return model def run_severity_agent(fnol_result: dict, coverage_result: dict, fraud_result: dict, submission: dict) -> dict: claim_id = fnol_result.get('claim_id', '') log.info(f"[SEVERITY] Agent 10 running for {claim_id}") feats = _build_features(fnol_result, coverage_result, fraud_result, submission) severity_band = 'MODERATE' reserve_estimate = 0.0 method = 'rules' try: import joblib model = joblib.load(MODEL_PATH) X = pd.DataFrame([feats])[model['feature_cols']] pred = model['classifier'].predict(X)[0] severity_band = model['label_encoder'].inverse_transform([pred])[0] reserve_estimate = float(np.clip(model['regressor'].predict(X)[0], 0, 5000000)) method = 'xgboost' except Exception as e: log.warning(f"[SEVERITY] ML unavailable: {e} — using rules") # Rules fallback ratio = feats['amount_ratio'] if ratio < 0.10: severity_band = 'MINOR' elif ratio < 0.35: severity_band = 'MODERATE' elif ratio < 0.70: severity_band = 'MAJOR' else: severity_band = 'CATASTROPHIC' net = float(coverage_result.get('net_payable_est') or 0) reserve_estimate = net * 0.82 # ── Reserve adjustment for fraud ───────────────────────── fraud_score = int(fraud_result.get('fraud_score') or 0) if fraud_score >= 70: reserve_estimate *= 0.5 # Hold lower reserve pending SIU reserve_note = f"Reserve reduced by 50% — high fraud score ({fraud_score})" elif fraud_score >= 40: reserve_estimate *= 0.75 reserve_note = f"Reserve reduced by 25% — moderate fraud score ({fraud_score})" else: reserve_note = "Full reserve applied" reserve_estimate = round(reserve_estimate, 2) # ── Severity summary ────────────────────────────────────── severity_desc = { 'MINOR': 'Low-value claim, likely straightforward settlement', 'MODERATE': 'Standard claim requiring adjuster review', 'MAJOR': 'Significant loss requiring senior adjuster and inspection', 'CATASTROPHIC': 'Total or near-total loss — executive escalation required', } result = { 'claim_id': claim_id, 'status': 'SEVERITY_SCORED', 'severity_band': severity_band, 'severity_desc': severity_desc.get(severity_band, ''), 'reserve_estimate': reserve_estimate, 'reserve_note': reserve_note, '_method': method, } log.info( f"[SEVERITY] {claim_id}: band={severity_band} " f"reserve=${reserve_estimate:,.0f} method={method}" ) return result