""" PolicyBridge — Flask API Server ================================ Run locally: python app.py (connects to localhost MySQL) Run on HF: Set Secrets below, app auto-detects and connects to Clever Cloud HuggingFace Secrets to add (Settings → Variables and Secrets): MYSQL_ADDON_HOST = btvbbpqhvnttzvptguj3-mysql.services.clever-cloud.com MYSQL_ADDON_PORT = 3306 MYSQL_ADDON_USER = utenclk29u394u1j MYSQL_ADDON_PASSWORD = QXFZTmUtPnXrKFqZKpLQ MYSQL_ADDON_DB = btvbbpqhvnttzvptguj3 """ import os, sys, json, logging from datetime import datetime from flask import Flask, jsonify, request, send_from_directory from flask_cors import CORS AGENTS_DIR = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, AGENTS_DIR) def _import_agent(module_name, func_name): try: mod = __import__(module_name) return getattr(mod, func_name) except Exception as e: logging.warning(f"Could not import {module_name}.{func_name}: {e}") return None # ════════════════════════════════════════════════════════════════════ # DYNAMIC ENVIRONMENT — Local MySQL vs HuggingFace + Clever Cloud # ════════════════════════════════════════════════════════════════════ 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: return os.environ.get(addon_key) or os.environ.get(generic_key) or default DB = dict( host = _env("MYSQL_ADDON_HOST", "MYSQL_HOST", "localhost"), port = int(_env("MYSQL_ADDON_PORT", "MYSQL_PORT", "3306")), user = _env("MYSQL_ADDON_USER", "MYSQL_USER", "root"), password = _env("MYSQL_ADDON_PASSWORD", "MYSQL_PASSWORD", "root@123"), database = _env("MYSQL_ADDON_DB", "MYSQL_DATABASE", "bronze"), ) def T(layer: str, table: str) -> str: return f"`{layer}_{table}`" if _is_huggingface() else f"`{layer}`.`{table}`" app = Flask(__name__) CORS(app, origins="*") logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') log = logging.getLogger(__name__) _engine = None def get_engine(): """ NullPool engine — NO connection pooling. Each request opens one connection and closes it immediately when done. This is the only safe approach for Clever Cloud free tier (max 5 connections) because pooling keeps connections open between requests. With NullPool: active connections = number of requests being processed RIGHT NOW. On a single-worker server this is almost always 1. """ global _engine if _engine is None: from sqlalchemy import create_engine from sqlalchemy.pool import NullPool from urllib.parse import quote_plus as qp pwd = qp(DB['password']) _engine = create_engine( f"mysql+pymysql://{DB['user']}:{pwd}@{DB['host']}:{DB['port']}/{DB['database']}?charset=utf8mb4", poolclass=NullPool, connect_args={"connect_timeout": 10} ) log.info(f"[DB] NullPool engine created → {DB['host']}:{DB['port']}/{DB['database']}") return _engine def get_conn(): return get_engine().connect() pipeline_ctx: dict = {} # ════════════════════════════════════════════════════════════════════ # HELPERS # ════════════════════════════════════════════════════════════════════ def safe_json(obj): import math, numpy as np if isinstance(obj, dict): return {k: safe_json(v) for k, v in obj.items()} if isinstance(obj, list): return [safe_json(v) for v in obj] if isinstance(obj, float): return None if (math.isnan(obj) or math.isinf(obj)) else obj if isinstance(obj, np.integer): return int(obj) if isinstance(obj, np.floating): v = float(obj); return None if (math.isnan(v) or math.isinf(v)) else v if isinstance(obj, np.bool_): return bool(obj) if isinstance(obj, np.ndarray): return obj.tolist() return obj def df_to_records(df) -> list: return df.where(df.notna(), other=None).to_dict(orient="records") def load_submission(sub_id: str) -> dict | None: import pandas as pd try: eng = get_engine() query = f""" SELECT s.submission_id, s.coverage_type_code, s.requested_coverage_limit, s.requested_deductible, s.pipeline_status, s.final_outcome, s.halt_reason, s.raw_payload, i.full_name, i.dob, i.email, i.phone, i.city AS insured_city, i.state_code AS insured_state, i.street AS insured_street, i.zip AS insured_zip, p.street AS prop_street, p.city AS prop_city, p.state_code, p.zip AS prop_zip, p.year_built, p.square_footage, p.construction_type, p.roof_type, p.roof_year, p.num_stories, p.property_type, p.occupancy, b.broker_code, b.broker_name FROM {T('bronze','submissions')} s LEFT JOIN {T('bronze','insureds')} i ON s.insured_id = i.insured_id LEFT JOIN {T('bronze','properties')} p ON s.property_id = p.property_id LEFT JOIN {T('bronze','brokers')} b ON s.broker_id = b.broker_id WHERE s.submission_id = %(sid)s LIMIT 1 """ with eng.connect() as conn: df = pd.read_sql(query, conn, params={"sid": sub_id}) if df.empty: return None row = df_to_records(df)[0] payload = {} raw_str = row.get("raw_payload") or "" if raw_str: try: payload = json.loads(raw_str) except json.JSONDecodeError: log.warning(f"load_submission({sub_id}): JSON corrupt — rebuilding from DB") try: import re clean = re.sub(r',\s*"[^"]*$', '', raw_str).rstrip(',') clean += '}' * max(clean.count('{') - clean.count('}'), 0) payload = json.loads(clean) except Exception: payload = {} ins = payload.setdefault("insured", {}) prop = payload.setdefault("property", {}) payload.setdefault("policy_request", {}) payload.setdefault("agent_results", {}) ins.update({k: v for k, v in { "full_name": row.get("full_name"), "dob": str(row.get("dob","")) if row.get("dob") else None, "email": row.get("email"), "phone": row.get("phone"), }.items() if v is not None}) prop.update({k: v for k, v in { "street": row.get("prop_street"), "city": row.get("prop_city"), "state": row.get("state_code"), "state_code": row.get("state_code"), "zip": row.get("prop_zip"), "year_built": row.get("year_built"), "square_footage": row.get("square_footage"), "construction_type": row.get("construction_type"), "roof_type": row.get("roof_type"), "roof_year": row.get("roof_year"), "num_stories": row.get("num_stories"), "property_type": row.get("property_type"), "occupancy": row.get("occupancy"), }.items() if v is not None}) if not ins.get("credit_score"): ins["credit_score"] = 650 if not ins.get("kyc_status"): ins["kyc_status"] = "PENDING" payload.update({ "_submission_id": sub_id, "_coverage_type_code": row.get("coverage_type_code"), "_requested_coverage_limit": row.get("requested_coverage_limit"), "_requested_deductible": row.get("requested_deductible"), "_pipeline_status": row.get("pipeline_status"), "_final_outcome": row.get("final_outcome"), "_broker_code": row.get("broker_code"), "_broker_name": row.get("broker_name"), }) log.info(f"load_submission({sub_id}): OK — {ins.get('full_name')} / {prop.get('city')}") return payload except Exception as e: log.error(f"load_submission({sub_id}): {e}", exc_info=True) return None # ════════════════════════════════════════════════════════════════════ # ROUTES # ════════════════════════════════════════════════════════════════════ @app.route("/status", methods=["GET"]) def status(): try: import pandas as pd conn = get_conn() df = pd.read_sql(f"SELECT COUNT(*) AS c FROM {T('bronze','submissions')}", conn) tbls = pd.read_sql("SHOW TABLES", conn).iloc[:, 0].tolist() conn.close() audit_table_exists = any("audit_log" in t for t in tbls) return jsonify({ "status": "ok", "environment": "huggingface" if _is_huggingface() else "local", "db_host": DB['host'], "db_name": DB['database'], "submissions_count": int(df_to_records(df)[0].get("c", 0)), "tables": tbls, "timestamp": datetime.now().isoformat(), "agents": { "agent1_kyc": os.path.exists(os.path.join(AGENTS_DIR,"models","agent1_kyc_classifier.pkl")), "agent2_property": os.path.exists(os.path.join(AGENTS_DIR,"models","agent2_property_risk.pkl")), "agent3_underwriting": os.path.exists(os.path.join(AGENTS_DIR,"models","agent3_underwriting.pkl")), "agent4_pricing": os.path.exists(os.path.join(AGENTS_DIR,"models","agent4_pricing.pkl")), "agent6_audit": True, }, "audit_log_table": audit_table_exists, }) except Exception as e: return jsonify({"status": "error", "error": str(e)}), 500 @app.route("/submissions/recent", methods=["GET"]) def recent_submissions(): limit = request.args.get("limit", 10, type=int) try: import pandas as pd eng = get_engine() query = f""" SELECT s.submission_id, s.coverage_type_code, s.requested_coverage_limit, s.requested_deductible, s.final_outcome, s.pipeline_status, s.halt_reason, s.submitted_at, s.raw_payload, i.full_name, p.city AS prop_city, p.state_code AS prop_state, p.street AS prop_street, p.zip AS prop_zip, p.year_built, p.square_footage, p.construction_type, p.roof_type, p.roof_year, p.num_stories, p.property_type, b.broker_code, b.broker_name FROM {T('bronze','submissions')} s LEFT JOIN {T('bronze','insureds')} i ON s.insured_id = i.insured_id LEFT JOIN {T('bronze','properties')} p ON s.property_id = p.property_id LEFT JOIN {T('bronze','brokers')} b ON s.broker_id = b.broker_id ORDER BY s.submitted_at DESC LIMIT {limit} """ with eng.connect() as conn: df = pd.read_sql(query, conn) subs = [] for d in df_to_records(df): try: payload = json.loads(d["raw_payload"]) if d.get("raw_payload") else {} except: payload = {} ins = payload.setdefault("insured", {}) prop = payload.setdefault("property", {}) payload.setdefault("policy_request", {}); payload.setdefault("agent_results", {}) if d.get("full_name"): ins["full_name"] = d["full_name"] if d.get("prop_city"): prop["city"] = d["prop_city"] if d.get("prop_state"): prop["state"] = d["prop_state"]; prop["state_code"] = d["prop_state"] if d.get("prop_street"): prop["street"] = d["prop_street"] if d.get("prop_zip"): prop["zip"] = d["prop_zip"] for fld in ["year_built","square_footage","construction_type","roof_type","roof_year","num_stories","property_type"]: if d.get(fld) is not None: prop[fld] = d[fld] subs.append(safe_json({ "submission_id": d.get("submission_id"), "coverage_type_code": d.get("coverage_type_code"), "requested_coverage_limit": d.get("requested_coverage_limit"), "requested_deductible": d.get("requested_deductible"), "final_outcome": d.get("final_outcome"), "pipeline_status": d.get("pipeline_status"), "halt_reason": d.get("halt_reason"), "submitted_at": str(d.get("submitted_at","")), "broker_code": d.get("broker_code"), "broker_name": d.get("broker_name"), "raw_payload": payload, })) return jsonify({"submissions": subs, "count": len(subs)}) except Exception as e: log.error(f"/submissions/recent: {e}", exc_info=True) return jsonify({"error": str(e), "submissions": [], "count": 0}), 500 @app.route("/debug/tables", methods=["GET"]) def debug_tables(): try: import pandas as pd eng = get_engine() with eng.connect() as c: tbls = pd.read_sql("SHOW TABLES", c).iloc[:, 0].tolist() result = {} for tbl in tbls: try: with eng.connect() as c: cols = pd.read_sql(f"SHOW COLUMNS FROM `{tbl}`", c) sample = pd.read_sql(f"SELECT * FROM `{tbl}` LIMIT 1", c) result[tbl] = {"columns": cols["Field"].tolist(), "sample": safe_json(sample.iloc[0].to_dict()) if not sample.empty else {}} except Exception as te: result[tbl] = {"error": str(te)} return jsonify(result) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/debug/submission", methods=["GET"]) def debug_submission(): try: import pandas as pd conn = get_conn() df = pd.read_sql(f"SELECT * FROM {T('bronze','submissions')} ORDER BY submitted_at DESC LIMIT 1", conn) conn.close() if df.empty: return jsonify({"error": "No submissions found"}) row = df.iloc[0].to_dict() try: row["raw_payload"] = json.loads(row["raw_payload"]) if row["raw_payload"] else {} except: pass for col in ["submitted_at","created_at","updated_at","received_at"]: row[col] = str(row.get(col,"")) return jsonify(safe_json(row)) except Exception as e: return jsonify({"error": str(e)}), 500 # ── AGENT 1: KYC ────────────────────────────────────────────────── @app.route("/agent/kyc", methods=["POST"]) def agent_kyc(): sub_id = request.json.get("submission_id") if not sub_id: return jsonify({"error": "submission_id required"}), 400 sub = pipeline_ctx.get(sub_id, {}).get("submission") or load_submission(sub_id) if not sub: return jsonify({"error": f"Submission {sub_id} not found"}), 404 log.info(f"[KYC] Running Agent 1 for {sub_id}") try: from agent1_ssn_kyc import run_kyc_agent result = run_kyc_agent(sub) except Exception as e: log.error(f"[KYC] Agent error: {e}") credit = (sub.get("insured") or {}).get("credit_score", 0) ofac = (sub.get("insured") or {}).get("ofac_result", "CLEAR") result = {"submission_id": sub_id, "status": "KYC_PASS" if credit >= 550 and ofac == "CLEAR" else "KYC_FAIL", "credit_score": credit, "ofac_result": ofac, "fraud_signals": 0, "ml_kyc_probability": 0.85 if credit >= 550 else 0.1, "decline_reason": None if credit >= 550 else f"Credit {credit} below 550", "_fallback": True} pipeline_ctx.setdefault(sub_id, {}).update({"kyc": result, "submission": sub}) return jsonify(safe_json(result)) # ── AGENT 2: PROPERTY RISK ──────────────────────────────────────── @app.route("/agent/property", methods=["POST"]) def agent_property(): sub_id = request.json.get("submission_id") if not sub_id: return jsonify({"error": "submission_id required"}), 400 ctx = pipeline_ctx.get(sub_id, {}) sub = ctx.get("submission") or load_submission(sub_id) kyc = ctx.get("kyc", {"status": "KYC_PASS"}) log.info(f"[PROPERTY] Running Agent 2 for {sub_id}") try: from agent2_property_risk import run_property_risk_agent result = run_property_risk_agent(kyc, sub) except Exception as e: log.error(f"[PROPERTY] Agent error: {e}") prop = (sub or {}).get("property", {}) score = prop.get("prop_risk_score", 40) band = prop.get("risk_band", "MEDIUM") or "MEDIUM" is_ok = score is not None and score < 75 result = {"submission_id": sub_id, "status": "RISK_ACCEPTABLE" if is_ok else "RISK_DECLINED", "risk_band": band if is_ok else "DECLINED", "peril_scores": {"wind_score": round((score or 40)*0.65), "flood_score": round((score or 40)*0.55), "fire_score": round((score or 40)*0.60), "overall_risk": score or 40}, "risk_acceptability_prob": 0.8 if is_ok else 0.1, "decline_reason": None if is_ok else f"Risk score {score} exceeds threshold", "_fallback": True} pipeline_ctx.setdefault(sub_id, {}).update({"property": result, "submission": sub}) return jsonify(safe_json(result)) # ── AGENT 3: UNDERWRITING ───────────────────────────────────────── @app.route("/agent/underwriting", methods=["POST"]) def agent_underwriting(): sub_id = request.json.get("submission_id") if not sub_id: return jsonify({"error": "submission_id required"}), 400 ctx = pipeline_ctx.get(sub_id, {}) sub = ctx.get("submission") or load_submission(sub_id) prop = ctx.get("property", {"status": "RISK_ACCEPTABLE", "peril_scores": {}}) log.info(f"[UW] Running Agent 3 for {sub_id}") try: from agent3_underwriting import run_underwriting_agent result = run_underwriting_agent(prop, sub) except Exception as e: log.error(f"[UW] Agent error: {e}") cov = (sub or {}).get("_coverage_type_code") or (sub or {}).get("policy_request",{}).get("coverage_type","HO-3") lim = (sub or {}).get("_requested_coverage_limit") or (sub or {}).get("policy_request",{}).get("limit",300000) result = {"submission_id": sub_id, "status": "UW_APPROVED", "coverage_type": cov, "uw_approval_probability": 0.82, "expected_loss_ratio": 0.58, "primary_decline_reason": None, "all_rule_violations": [], "reinsurance_required": lim > 2_000_000, "_fallback": True} pipeline_ctx.setdefault(sub_id, {}).update({"underwriting": result}) return jsonify(safe_json(result)) # ── AGENT 4: PRICING ────────────────────────────────────────────── @app.route("/agent/pricing", methods=["POST"]) def agent_pricing(): sub_id = request.json.get("submission_id") if not sub_id: return jsonify({"error": "submission_id required"}), 400 ctx = pipeline_ctx.get(sub_id, {}) sub = ctx.get("submission") or load_submission(sub_id) uw = ctx.get("underwriting", {"status": "UW_APPROVED"}) prop = ctx.get("property", {"peril_scores": {}}) log.info(f"[PRICING] Running Agent 4 for {sub_id}") try: from agent4_pricing import run_pricing_agent result = run_pricing_agent(uw, prop, sub) except Exception as e: log.error(f"[PRICING] Agent error: {e}") lim = float((sub or {}).get("_requested_coverage_limit") or (sub or {}).get("policy_request",{}).get("limit", 300000)) prem = max(300, round(lim * 0.0065)) result = {"submission_id": sub_id, "final_premium": prem, "actuarial_premium": prem, "annual_premium": prem, "monthly_premium": round(prem/12), "confidence_interval": {"lo_95": round(prem*0.88), "hi_95": round(prem*1.12)}, "premium_breakdown": {"credit_modifier":1.0,"risk_modifier":1.0,"age_modifier":1.0}, "_fallback": True} pipeline_ctx.setdefault(sub_id, {}).update({"pricing": result}) return jsonify(safe_json(result)) # ── AGENT 5: ISSUANCE ───────────────────────────────────────────── @app.route("/agent/issuance", methods=["POST"]) def agent_issuance(): sub_id = request.json.get("submission_id") if not sub_id: return jsonify({"error": "submission_id required"}), 400 ctx = pipeline_ctx.get(sub_id, {}) sub = ctx.get("submission") or load_submission(sub_id) pric = ctx.get("pricing", {"final_premium": 0}) uw = ctx.get("underwriting", {}) prop = ctx.get("property", {}) log.info(f"[ISSUANCE] Running Agent 5 for {sub_id}") try: from agent5_issuance_orchestrator import run_issuance_agent result = run_issuance_agent(pric, uw, prop, sub) except Exception as e: log.error(f"[ISSUANCE] Agent error: {e}") yr = datetime.now().year prem = pric.get("final_premium", 0) result = {"submission_id": sub_id, "policy_number": f"PC-{yr}-{sub_id[-5:]}{(sub or {}).get('property',{}).get('state','XX')}", "policy_details": { "effective_date": datetime.now().strftime("%Y-%m-%d"), "expiration_date": f"{yr+1}-{datetime.now().strftime('%m-%d')}", "annual_premium": prem, "monthly_premium": round(prem/12), "coverage_type": uw.get("coverage_type","HO-3"), "risk_band": prop.get("risk_band","MEDIUM"), }, "documents_generated": ["declarations_page.pdf","policy_contract.pdf"], "notifications_sent": {"email":True,"sns":True}, "_fallback": True} # Store issuance in ctx BEFORE popping (audit needs it) pipeline_ctx.setdefault(sub_id, {}).update({"issuance": result}) _update_bronze_status(sub_id, result) return jsonify(safe_json(result)) # ════════════════════════════════════════════════════════════════════ # AGENT 6: AUDIT — AXIOM # Reads all prior agent outputs from pipeline_ctx (or DB fallback) # Generates plain-English audit summary + saves to gold_audit_log # ════════════════════════════════════════════════════════════════════ def _run_audit_from_ctx(sub_id: str, ctx: dict) -> dict: """Core audit logic — works from in-memory ctx or loaded submission.""" sub = ctx.get("submission", {}) kyc = ctx.get("kyc", {}) prop = ctx.get("property", {}) uw = ctx.get("underwriting", {}) pric = ctx.get("pricing", {}) iss = ctx.get("issuance", {}) factors = [] decline_reasons = [] # ── Agent 1: KYC / Document Validation ──────────────────────── kyc_pass = kyc.get("status") == "KYC_PASS" credit = kyc.get("credit_score", 0) ofac = kyc.get("ofac_result", "CLEAR") fraud_sig = kyc.get("fraud_signals", 0) kyc_prob = float(kyc.get("ml_kyc_probability") or 0.85) if kyc: factors.append({ "agent": "Document Validation Agent", "factor": "Identity & Compliance Check", "outcome": "PASS" if kyc_pass else "FAIL", "detail": ( f"Credit score: {credit}. " f"OFAC: {'Clear' if ofac == 'CLEAR' else 'HIT — FLAGGED'}. " f"Fraud signals: {fraud_sig}. " f"KYC model confidence: {kyc_prob:.1%}." ) }) if not kyc_pass: reason = kyc.get("decline_reason") or f"Credit {credit} below minimum or OFAC flag detected." decline_reasons.append(f"Document Validation failed: {reason}") # ── Agent 2: Property Risk ───────────────────────────────────── risk_band = prop.get("risk_band", "MEDIUM") peril = prop.get("peril_scores", {}) wind = peril.get("wind_score", 0) flood = peril.get("flood_score", 0) fire = peril.get("fire_score", 0) overall_r = peril.get("overall_risk", 0) risk_ok = prop.get("status") != "RISK_DECLINED" if prop: factors.append({ "agent": "Property Risk Agent", "factor": "Peril Risk Assessment", "outcome": "PASS" if risk_ok else "DECLINE", "detail": ( f"Risk band: {risk_band}. " f"Wind: {wind} | Flood: {flood} | Fire: {fire}. " f"Overall risk score: {overall_r}." ) }) if not risk_ok: reason = prop.get("decline_reason") or f"Risk score {overall_r} exceeds threshold." decline_reasons.append(f"Property risk declined: {reason}") # ── Agent 3: Underwriting ────────────────────────────────────── uw_status = uw.get("status", "UW_APPROVED") uw_approved = uw_status == "UW_APPROVED" uw_prob = float(uw.get("uw_approval_probability") or 0.82) uw_elr = float(uw.get("expected_loss_ratio") or 0.58) uw_reason = uw.get("primary_decline_reason") or "" violations = uw.get("all_rule_violations", []) if uw: factors.append({ "agent": "Underwriting Agent", "factor": "AI Underwriting Decision", "outcome": "APPROVED" if uw_approved else "DECLINED", "detail": ( f"Decision: {uw_status}. " f"Approval probability: {uw_prob:.1%}. " f"Expected loss ratio: {uw_elr:.2f}. " f"Rule violations: {len(violations)}." + (f" Decline reason: {uw_reason}." if uw_reason else "") ) }) if not uw_approved: decline_reasons.append( f"Underwriting declined: {uw_reason or 'Risk profile outside binding authority guidelines'}. " f"Approval probability: {uw_prob:.1%}." ) # ── Agent 4: Pricing ────────────────────────────────────────── final_prem = float(pric.get("final_premium") or pric.get("annual_premium") or 0) monthly_prem = float(pric.get("monthly_premium") or (final_prem / 12 if final_prem else 0)) ci = pric.get("confidence_interval", {}) breakdown = pric.get("premium_breakdown", {}) credit_mod = float(breakdown.get("credit_modifier") or 1.0) risk_mod = float(breakdown.get("risk_modifier") or 1.0) if pric and final_prem > 0: factors.append({ "agent": "Pricing Agent", "factor": "Actuarial Premium Calculation", "outcome": "CALCULATED", "detail": ( f"Annual premium: ${final_prem:,.2f} (${monthly_prem:,.2f}/mo). " f"Credit modifier: {credit_mod:.2f}x | Risk modifier: {risk_mod:.2f}x. " f"95% CI: ${ci.get('lo_95', 0):,.0f} – ${ci.get('hi_95', 0):,.0f}." ) }) # ── Agent 5: Issuance ───────────────────────────────────────── policy_num = iss.get("policy_number") pol_details = iss.get("policy_details", {}) eff_date = pol_details.get("effective_date", "") exp_date = pol_details.get("expiration_date", "") docs = iss.get("documents_generated", []) if iss: factors.append({ "agent": "Issuance Agent", "factor": "Policy Issuance", "outcome": "ISSUED" if policy_num else "NOT ISSUED", "detail": ( f"Policy number: {policy_num or 'N/A'}. " f"Effective: {eff_date} to {exp_date}. " f"Documents: {', '.join(docs) if docs else 'None generated'}." ) }) # ── Final decision ───────────────────────────────────────────── decision = "DECLINED" if decline_reasons else "APPROVED" ins_name = (sub.get("insured") or {}).get("full_name", "the applicant") prop_city = (sub.get("property") or {}).get("city", "") prop_state = (sub.get("property") or {}).get("state_code", "") location = f"{prop_city}, {prop_state}".strip(", ") or "the insured property" if decision == "APPROVED": overall_summary = ( f"The application from {ins_name} for the property at {location} has been approved. " f"The property was assessed as {risk_band} risk across all perils, " f"identity and compliance screening passed with {fraud_sig} fraud signal(s) detected, " f"and the underwriting model approved the submission at {uw_prob:.1%} confidence. " f"A final annual premium of ${final_prem:,.2f} has been calculated and policy " f"{policy_num} has been issued." ) else: reason_text = " ".join(decline_reasons[:2]) overall_summary = ( f"The application from {ins_name} for the property at {location} has been declined. " f"{reason_text} " f"All {len(factors)} agent checks were completed before reaching this decision." ) # Composite audit confidence score (0–100) score_parts = [] if kyc: score_parts.append(100 if kyc_pass else 0) if prop: score_parts.append({"LOW":100,"MEDIUM":70,"HIGH":30,"DECLINED":0}.get(risk_band, 50)) if uw: score_parts.append(int(uw_prob * 100)) audit_score = int(sum(score_parts) / len(score_parts)) if score_parts else 50 audit_result = { "submission_id": sub_id, "decision": decision, "overall_summary": overall_summary, "decision_factors": factors, "decline_reasons": decline_reasons, "risk_band": risk_band, "final_premium": final_prem if final_prem > 0 else None, "policy_number": policy_num, "audit_score": audit_score, "audited_at": datetime.utcnow().isoformat(), } # ── Save to gold_audit_log ───────────────────────────────────── try: from sqlalchemy import text as sqlt eng = get_engine() with eng.begin() as conn: r = conn.execute(sqlt(f""" INSERT INTO {T('gold','audit_log')} ( submission_id, decision, overall_summary, decision_factors_json, decline_reasons_json, risk_band, final_premium, policy_number, audit_score, audited_at ) VALUES ( :submission_id, :decision, :overall_summary, :factors_json, :reasons_json, :risk_band, :final_premium, :policy_number, :audit_score, :audited_at ) """), { "submission_id": sub_id, "decision": decision, "overall_summary": overall_summary, "factors_json": json.dumps(factors), "reasons_json": json.dumps(decline_reasons), "risk_band": risk_band, "final_premium": final_prem if final_prem > 0 else None, "policy_number": policy_num, "audit_score": audit_score, "audited_at": datetime.utcnow(), }) audit_result["audit_log_id"] = r.lastrowid log.info(f"[AUDIT] Saved audit_log_id={audit_result['audit_log_id']} for {sub_id} — {decision}") except Exception as db_err: log.warning(f"[AUDIT] Could not save to gold_audit_log: {db_err}") audit_result["audit_log_id"] = None audit_result["_db_warning"] = str(db_err) return audit_result @app.route("/agent/audit", methods=["POST"]) def agent_audit(): """ Agent 6 — AXIOM Audit Agent. POST /agent/audit Body: {"submission_id": "SUB-001"} Generates plain-English audit summary and saves to gold_audit_log. """ sub_id = request.json.get("submission_id") if not sub_id: return jsonify({"error": "submission_id required"}), 400 log.info(f"[AUDIT] Running Agent 6 (AXIOM) for {sub_id}") ctx = pipeline_ctx.get(sub_id, {}) if not ctx.get("submission"): sub = load_submission(sub_id) if not sub: return jsonify({"error": f"Submission {sub_id} not found"}), 404 ctx["submission"] = sub result = _run_audit_from_ctx(sub_id, ctx) return jsonify(safe_json(result)) @app.route("/audit/", methods=["GET"]) def get_audit(sub_id): """ GET /audit/ Returns the most recent audit record from gold_audit_log. Runs audit on demand if no record exists yet. """ try: import pandas as pd eng = get_engine() with eng.connect() as conn: df = pd.read_sql( f"SELECT * FROM {T('gold','audit_log')} WHERE submission_id = %(sid)s ORDER BY audited_at DESC LIMIT 1", conn, params={"sid": sub_id} ) if not df.empty: record = df_to_records(df)[0] for col in ["decision_factors_json", "decline_reasons_json"]: val = record.get(col) if val: try: record[col.replace("_json", "")] = json.loads(val) except Exception: record[col.replace("_json", "")] = [] record["audited_at"] = str(record.get("audited_at", "")) return jsonify(safe_json(record)) # Not in DB — run on demand log.info(f"[AUDIT] No record for {sub_id}, running on demand") ctx = pipeline_ctx.get(sub_id, {}) if not ctx.get("submission"): sub = load_submission(sub_id) if not sub: return jsonify({"error": f"Submission {sub_id} not found"}), 404 ctx["submission"] = sub return jsonify(safe_json(_run_audit_from_ctx(sub_id, ctx))) except Exception as e: log.error(f"[AUDIT] GET /audit/{sub_id}: {e}", exc_info=True) return jsonify({"error": str(e)}), 500 @app.route("/audit/run/", methods=["POST"]) def trigger_audit(sub_id): """ POST /audit/run/ Manually re-trigger audit for any existing submission (backfill). """ log.info(f"[AUDIT] Manual trigger for {sub_id}") ctx = pipeline_ctx.get(sub_id, {}) if not ctx.get("submission"): sub = load_submission(sub_id) if not sub: return jsonify({"error": f"Submission {sub_id} not found"}), 404 ctx["submission"] = sub return jsonify(safe_json(_run_audit_from_ctx(sub_id, ctx))) # ── FULL PIPELINE (Agents 1–6) ──────────────────────────────────── @app.route("/pipeline/run", methods=["POST"]) def pipeline_run(): sub_id = request.json.get("submission_id") if not sub_id: return jsonify({"error": "submission_id required"}), 400 sub = load_submission(sub_id) if not sub: return jsonify({"error": f"Submission {sub_id} not found"}), 404 pipeline_ctx[sub_id] = {"submission": sub} steps = {} def call(ep): with app.test_client() as c: r = c.post(f"/agent/{ep}", json={"submission_id": sub_id}, content_type="application/json") return json.loads(r.data) # Agent 1: KYC steps["kyc"] = call("kyc") if steps["kyc"].get("status") != "KYC_PASS": steps["audit"] = call("audit") return jsonify({ "final_outcome": "DECLINED", "halt_reason": "KYC_FAIL", "steps": steps, "audit_summary": steps["audit"].get("overall_summary"), "audit_score": steps["audit"].get("audit_score"), "audit_log_id": steps["audit"].get("audit_log_id"), }) # Agent 2: Property Risk steps["property"] = call("property") if steps["property"].get("status") == "RISK_DECLINED": steps["audit"] = call("audit") return jsonify({ "final_outcome": "DECLINED", "halt_reason": "PROP_DECLINED", "steps": steps, "audit_summary": steps["audit"].get("overall_summary"), "audit_score": steps["audit"].get("audit_score"), "audit_log_id": steps["audit"].get("audit_log_id"), }) # Agent 3: Underwriting steps["underwriting"] = call("underwriting") if steps["underwriting"].get("status") == "UW_DECLINED": steps["audit"] = call("audit") return jsonify({ "final_outcome": "DECLINED", "halt_reason": "UW_DECLINED", "steps": steps, "audit_summary": steps["audit"].get("overall_summary"), "audit_score": steps["audit"].get("audit_score"), "audit_log_id": steps["audit"].get("audit_log_id"), }) # Agent 4: Pricing steps["pricing"] = call("pricing") # Agent 5: Issuance steps["issuance"] = call("issuance") # Agent 6: Audit — always runs at pipeline end steps["audit"] = call("audit") return jsonify({ "final_outcome": "APPROVED", "policy_number": steps["issuance"].get("policy_number"), "final_premium": steps["pricing"].get("final_premium"), "audit_summary": steps["audit"].get("overall_summary"), "audit_score": steps["audit"].get("audit_score"), "audit_log_id": steps["audit"].get("audit_log_id"), "steps": steps, }) # ════════════════════════════════════════════════════════════════════ # INTERNAL # ════════════════════════════════════════════════════════════════════ def _update_bronze_status(sub_id: str, issuance_result: dict): try: from sqlalchemy import text eng = get_engine() prem = (issuance_result.get("policy_details") or {}).get("annual_premium", 0) with eng.begin() as conn: conn.execute(text(f""" UPDATE {T('bronze','submissions')} SET pipeline_status = 'ISSUED', final_outcome = 'APPROVED', updated_at = NOW() WHERE submission_id = :sid """), {"sid": sub_id}) log.info(f"[DB] Updated {sub_id} → ISSUED / ${prem}") except Exception as e: log.warning(f"[DB] Could not update {sub_id}: {e}") # ════════════════════════════════════════════════════════════════════ # SUBMIT # ════════════════════════════════════════════════════════════════════ @app.route("/submit", methods=["POST"]) def submit_new(): try: from sqlalchemy import text data = request.get_json(force=True) if not data: return jsonify({"error": "No JSON payload received"}), 400 j = data.get("submission", data) sub = j if "submission_id" in j else data.get("submission", {}) if not sub: return jsonify({"error": "Missing 'submission' object"}), 400 sid = sub.get("submission_id", "") broker = sub.get("broker", {}) insured = sub.get("insured", {}) prop = sub.get("property", {}) pol_req = sub.get("policy_request", {}) attachments = sub.get("attachments", []) if not sid: return jsonify({"error": "Missing submission_id"}), 400 eng = get_engine() with eng.begin() as conn: # 1. BROKERS conn.execute(text(f""" INSERT INTO {T('bronze','brokers')} (broker_code, broker_name, contact_email, state_code) VALUES (:broker_code, :broker_name, :contact_email, :state_code) ON DUPLICATE KEY UPDATE broker_name=VALUES(broker_name), contact_email=VALUES(contact_email), state_code=VALUES(state_code) """), {"broker_code": broker.get("broker_code",""), "broker_name": broker.get("name",""), "contact_email": broker.get("contact_email",""), "state_code": broker.get("state_code","")}) broker_row = conn.execute(text( f"SELECT broker_id FROM {T('bronze','brokers')} WHERE broker_code = :bc" ), {"bc": broker.get("broker_code","")}).fetchone() broker_id = broker_row[0] if broker_row else None # 2. INSUREDS import hashlib ssn_raw = insured.get("ssn","") ssn_hash = hashlib.sha256(ssn_raw.encode()).hexdigest() if ssn_raw else None ins_result = conn.execute(text(f""" INSERT INTO {T('bronze','insureds')} (full_name, dob, ssn_hash, email, phone, street, city, state_code, zip) VALUES (:full_name, :dob, :ssn_hash, :email, :phone, :street, :city, :state_code, :zip) """), {"full_name": insured.get("full_name",""), "dob": insured.get("dob",None), "ssn_hash": ssn_hash, "email": insured.get("email",""), "phone": insured.get("phone",""), "street": insured.get("street",""), "city": insured.get("city",""), "state_code": insured.get("state",""), "zip": insured.get("zip","")}) insured_id = ins_result.lastrowid # 3. PROPERTIES prop_result = conn.execute(text(f""" INSERT INTO {T('bronze','properties')} (insured_id, street, city, state_code, zip, year_built, square_footage, construction_type, roof_type, roof_year, num_stories, property_type, occupancy) VALUES (:insured_id, :street, :city, :state_code, :zip, :year_built, :square_footage, :construction_type, :roof_type, :roof_year, :num_stories, :property_type, :occupancy) """), {"insured_id": insured_id, "street": prop.get("street",""), "city": prop.get("city",""), "state_code": prop.get("state_code", prop.get("state","")), "zip": prop.get("zip",""), "year_built": prop.get("year_built",None), "square_footage": prop.get("square_footage",None), "construction_type": prop.get("construction_type",""), "roof_type": prop.get("roof_type",""), "roof_year": prop.get("roof_year",None), "num_stories": prop.get("num_stories",None), "property_type": prop.get("property_type",""), "occupancy": prop.get("occupancy","")}) property_id = prop_result.lastrowid # 4. SUBMISSIONS import json as _json conn.execute(text(f""" INSERT INTO {T('bronze','submissions')} (submission_id, broker_id, insured_id, property_id, coverage_type_code, requested_coverage_limit, requested_deductible, line_of_business, market_type, submitted_at, received_at, pipeline_status, final_outcome, halt_reason, raw_payload) VALUES (:submission_id, :broker_id, :insured_id, :property_id, :coverage_type_code, :requested_coverage_limit, :requested_deductible, :line_of_business, :market_type, :submitted_at, NOW(), 'RECEIVED', NULL, NULL, :raw_payload) ON DUPLICATE KEY UPDATE pipeline_status='RECEIVED', received_at=NOW() """), {"submission_id": sid, "broker_id": broker_id, "insured_id": insured_id, "property_id": property_id, "coverage_type_code": pol_req.get("coverage_type",""), "requested_coverage_limit": pol_req.get("requested_coverage_limit",None), "requested_deductible": pol_req.get("requested_deductible",None), "line_of_business": sub.get("line_of_business",""), "market_type": sub.get("market_type",""), "submitted_at": sub.get("submitted_at",None), "raw_payload": _json.dumps(data)}) # 5. ATTACHMENTS for att in attachments: conn.execute(text(f""" INSERT IGNORE INTO {T('bronze','submission_attachments')} (submission_id, attachment_type, file_name, s3_uri, file_format, file_size_bytes, uploaded_at) VALUES (:submission_id, :attachment_type, :file_name, :s3_uri, :file_format, :file_size_bytes, NOW()) """), {"submission_id": sid, "attachment_type": att.get("attachment_type","OTHER"), "file_name": att.get("file_name",""), "s3_uri": att.get("s3_uri", f"s3://pcins-bronze/attachments/{sid}/{att.get('file_name','')}"), "file_format": att.get("file_format",""), "file_size_bytes": att.get("file_size_bytes",0)}) # 6. PIPELINE AGENT LOG — 6 agents including AXIOM for agent_code, seq in [ ("SSN_Identity_Validation_Agent", 1), ("Property_Risk_Assessment_Agent", 2), ("Underwriting_Decision_Agent", 3), ("Premium_Pricing_Agent", 4), ("Issuance_Agent", 5), ("Audit_Agent", 6), ]: conn.execute(text(f""" INSERT IGNORE INTO {T('bronze','pipeline_agent_log')} (submission_id, agent_code, agent_sequence, status, halt_reason, created_at) VALUES (:sid, :agent_code, :seq, 'QUEUED', 'Awaiting prior agents', NOW()) """), {"sid": sid, "agent_code": agent_code, "seq": seq}) log.info(f"[SUBMIT] {sid} inserted — insured_id={insured_id}, property_id={property_id}") return jsonify({"success": True, "submission_id": sid, "insured_id": insured_id, "property_id": property_id, "broker_id": broker_id, "message": f"{sid} committed to bronze · 6 agents queued"}) except Exception as e: log.error(f"[SUBMIT] Error: {e}", exc_info=True) return jsonify({"error": str(e)}), 500 # ════════════════════════════════════════════════════════════════════ # STATIC FILE SERVING — HTML front-end apps # ════════════════════════════════════════════════════════════════════ @app.route("/ui") @app.route("/ui/simulation") def simulation(): """Serve the Policy Simulation HTML — inline response, bypasses HF proxy.""" try: html = open(os.path.join(AGENTS_DIR, "pc_insurance_multiagent_v4.html"), encoding="utf-8").read() return html, 200, {"Content-Type": "text/html; charset=utf-8"} except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/ui/policybridge") def policybridge(): """Serve the PolicyBridge HTML — inline response.""" try: html = open(os.path.join(AGENTS_DIR, "policybridge_app.html"), encoding="utf-8").read() return html, 200, {"Content-Type": "text/html; charset=utf-8"} except Exception as e: return jsonify({"error": str(e)}), 500 # ════════════════════════════════════════════════════════════════════ # RISKRADAR — GEO RISK INTELLIGENCE (Agent 12) # ════════════════════════════════════════════════════════════════════ @app.route("/riskradar/score", methods=["POST"]) def riskradar_score(): """ Score a property address for wind/flood/fire/quake risk. POST /riskradar/score Body: { address, state_code, zip, year_built, construction_type, roof_year, num_stories, [latitude], [longitude], [submission_id] } """ try: from agent12_riskradar import run_riskradar_agent data = request.get_json(force=True) or {} if not data.get('state_code') or not data.get('zip'): return jsonify({"error": "state_code and zip are required"}), 400 result = run_riskradar_agent(data) # Persist to DB try: from sqlalchemy import text as sqlt eng = get_engine() with eng.begin() as conn: conn.execute(sqlt(""" INSERT INTO silver_riskradar_lookups ( address, state_code, zip_code, latitude, longitude, wind_score, flood_score, fire_score, quake_score, overall_score, overall_band, wind_band, flood_band, fire_band, quake_band, narrative, raw_result_json, source, submission_id ) VALUES ( :address, :state, :zip, :lat, :lng, :wind, :flood, :fire, :quake, :overall, :overall_band, :wind_band, :flood_band, :fire_band, :quake_band, :narrative, :raw, :source, :sub_id ) """), { "address": result.get("address"), "state": result["state_code"], "zip": result["zip_code"], "lat": result["latitude"], "lng": result["longitude"], "wind": result["scores"]["wind"], "flood": result["scores"]["flood"], "fire": result["scores"]["fire"], "quake": result["scores"]["quake"], "overall": result["overall_score"], "overall_band": result["overall_band"], "wind_band": result["bands"]["wind"], "flood_band": result["bands"]["flood"], "fire_band": result["bands"]["fire"], "quake_band": result["bands"]["quake"], "narrative": result["narrative"], "raw": json.dumps(safe_json(result)), "source": data.get("source", "LOOKUP"), "sub_id": data.get("submission_id"), }) except Exception as db_err: log.warning(f"[RISKRADAR] DB persist failed: {db_err}") return jsonify(safe_json(result)) except Exception as e: log.error(f"[RISKRADAR] /riskradar/score: {e}", exc_info=True) return jsonify({"error": str(e)}), 500 @app.route("/riskradar/history", methods=["GET"]) def riskradar_history(): """Recent risk lookups — last N records.""" limit = request.args.get("limit", 12, type=int) try: import pandas as pd eng = get_engine() with eng.connect() as conn: df = pd.read_sql( f"""SELECT lookup_id, address, state_code, zip_code, wind_score, flood_score, fire_score, quake_score, overall_score, overall_band, latitude, longitude, looked_up_at FROM silver_riskradar_lookups ORDER BY looked_up_at DESC LIMIT {limit}""", conn ) records = df.where(df.notna(), other=None).to_dict(orient="records") for r in records: r["looked_up_at"] = str(r.get("looked_up_at","")) return jsonify({"lookups": records, "count": len(records)}) except Exception as e: return jsonify({"error": str(e), "lookups": [], "count": 0}), 500 @app.route("/ui/riskradar") def riskradar_ui(): """Serve RiskRadar as standalone page.""" try: html = open(os.path.join(AGENTS_DIR, "pc_insurance_multiagent_v4.html"), encoding="utf-8").read() return html, 200, {"Content-Type": "text/html; charset=utf-8"} except Exception as e: return jsonify({"error": str(e)}), 500 # ════════════════════════════════════════════════════════════════════ # POLICYBRIDGE UI — SUBMISSION + MGA ONBOARDING ROUTES # ════════════════════════════════════════════════════════════════════ @app.route("/policybridge/submit", methods=["POST"]) def policybridge_submit(): try: from sqlalchemy import text as sqlt import hashlib data = request.get_json(force=True) or {} sub_id = data.get("submission_id", "") if not sub_id: return jsonify({"error": "submission_id required"}), 400 broker = data.get("broker", {}) insured = data.get("insured", {}) prop = data.get("property", {}) policy = data.get("policy_request", {}) mga = data.get("mga", {}) eng = get_engine() with eng.begin() as conn: # 1. Upsert broker conn.execute(sqlt(""" INSERT INTO bronze_brokers (broker_code, broker_name, contact_email, state_code) VALUES (:code, :name, :email, :state) ON DUPLICATE KEY UPDATE broker_name=VALUES(broker_name), contact_email=VALUES(contact_email), updated_at=NOW() """), { "code": broker.get("broker_code", ""), "name": broker.get("broker_name", ""), "email": broker.get("contact_email", ""), "state": broker.get("state_code", ""), }) broker_row = conn.execute(sqlt( "SELECT broker_id FROM bronze_brokers WHERE broker_code=:bc LIMIT 1" ), {"bc": broker.get("broker_code", "")}).fetchone() broker_id = broker_row[0] if broker_row else None # 2. Insert insured ssn_hash = hashlib.sha256( (insured.get("ssn") or sub_id).encode() ).hexdigest() ins_r = conn.execute(sqlt(""" INSERT INTO bronze_insureds (full_name, dob, ssn_hash, email, phone, street, city, state_code, zip) VALUES (:full_name, :dob, :ssn_hash, :email, :phone, :street, :city, :state_code, :zip) """), { "full_name": insured.get("full_name", ""), "dob": insured.get("dob") or None, "ssn_hash": ssn_hash, "email": insured.get("email", ""), "phone": insured.get("phone", ""), "street": insured.get("street", ""), "city": insured.get("city", ""), "state_code": insured.get("state", insured.get("state_code", "")), "zip": insured.get("zip", ""), }) insured_id = ins_r.lastrowid # 3. Insert property prop_r = conn.execute(sqlt(""" INSERT INTO bronze_properties (insured_id, street, city, state_code, zip, property_type, year_built, square_footage, construction_type, roof_type, roof_year, num_stories, occupancy) VALUES (:insured_id, :street, :city, :state_code, :zip, :property_type, :year_built, :sqft, :construction, :roof_type, :roof_year, :stories, :occupancy) """), { "insured_id": insured_id, "street": prop.get("street", ""), "city": prop.get("city", ""), "state_code": prop.get("state_code", prop.get("state", "")), "zip": prop.get("zip", ""), "property_type": prop.get("property_type", ""), "year_built": prop.get("year_built") or None, "sqft": prop.get("square_footage") or None, "construction": prop.get("construction_type", ""), "roof_type": prop.get("roof_type", ""), "roof_year": prop.get("roof_year") or None, "stories": prop.get("num_stories") or None, "occupancy": prop.get("occupancy", ""), }) property_id = prop_r.lastrowid # 4. Insert into policybridge_submissions conn.execute(sqlt(""" INSERT INTO policybridge_submissions ( submission_id, broker_code, broker_name, broker_email, broker_state, mga_code, insured_full_name, insured_dob, insured_email, insured_phone, insured_credit_score, insured_street, insured_city, insured_state, insured_zip, prop_street, prop_city, prop_state, prop_zip, prop_type, year_built, square_footage, construction_type, roof_type, roof_year, coverage_type_code, coverage_limit, deductible, effective_date, notes, pipeline_status, raw_payload_json, submitted_at ) VALUES ( :sub_id, :broker_code, :broker_name, :broker_email, :broker_state, :mga_code, :ins_name, :ins_dob, :ins_email, :ins_phone, :credit, :ins_street, :ins_city, :ins_state, :ins_zip, :prop_street, :prop_city, :prop_state, :prop_zip, :prop_type, :year_built, :sqft, :construction, :roof_type, :roof_year, :cov_type, :cov_limit, :deductible, :eff_date, :notes, 'RECEIVED', :raw, NOW() ) ON DUPLICATE KEY UPDATE pipeline_status='RECEIVED', updated_at=NOW() """), { "sub_id": sub_id, "broker_code": broker.get("broker_code", ""), "broker_name": broker.get("broker_name", ""), "broker_email": broker.get("contact_email", ""), "broker_state": broker.get("state_code", ""), "mga_code": mga.get("master_broker_code"), "ins_name": insured.get("full_name", ""), "ins_dob": insured.get("dob") or None, "ins_email": insured.get("email", ""), "ins_phone": insured.get("phone", ""), "credit": insured.get("credit_score"), "ins_street": insured.get("street", ""), "ins_city": insured.get("city", ""), "ins_state": insured.get("state", ""), "ins_zip": insured.get("zip", ""), "prop_street": prop.get("street", ""), "prop_city": prop.get("city", ""), "prop_state": prop.get("state_code", prop.get("state", "")), "prop_zip": prop.get("zip", ""), "prop_type": prop.get("property_type", ""), "year_built": prop.get("year_built"), "sqft": prop.get("square_footage"), "construction": prop.get("construction_type", ""), "roof_type": prop.get("roof_type", ""), "roof_year": prop.get("roof_year"), "cov_type": policy.get("coverage_type", ""), "cov_limit": policy.get("requested_coverage_limit"), "deductible": policy.get("requested_deductible"), "eff_date": policy.get("effective_date") or None, "notes": policy.get("notes", ""), "raw": json.dumps(data), }) # 5. Insert into bronze_submissions for pipeline conn.execute(sqlt(""" INSERT INTO bronze_submissions ( submission_id, broker_id, insured_id, property_id, coverage_type_code, requested_coverage_limit, requested_deductible, s3_uri, file_format, submitted_at, received_at, pipeline_status, raw_payload ) VALUES ( :sub_id, :broker_id, :insured_id, :property_id, :cov_type, :cov_limit, :deductible, :s3_uri, 'JSON', NOW(), NOW(), 'RECEIVED', :raw ) ON DUPLICATE KEY UPDATE pipeline_status='RECEIVED', updated_at=NOW() """), { "sub_id": sub_id, "broker_id": broker_id, "insured_id": insured_id, "property_id": property_id, "cov_type": policy.get("coverage_type", ""), "cov_limit": policy.get("requested_coverage_limit"), "deductible": policy.get("requested_deductible"), "s3_uri": f"s3://pcins-bronze/submissions/{sub_id}.json", "raw": json.dumps(data), }) # 6. Queue pipeline agents for agent_code, seq in [ ("SSN_Identity_Validation_Agent", 1), ("Property_Risk_Assessment_Agent", 2), ("Underwriting_Decision_Agent", 3), ("Premium_Pricing_Agent", 4), ("Issuance_Agent", 5), ("Audit_Agent", 6), ]: conn.execute(sqlt(""" INSERT IGNORE INTO bronze_pipeline_agent_log (submission_id, agent_code, agent_sequence, status, created_at) VALUES (:sid, :code, :seq, 'QUEUED', NOW()) """), {"sid": sub_id, "code": agent_code, "seq": seq}) log.info(f"[PB] {sub_id} written — insured_id={insured_id} property_id={property_id}") return jsonify({ "success": True, "submission_id": sub_id, "insured_id": insured_id, "property_id": property_id, "broker_id": broker_id, "message": f"{sub_id} committed to all bronze tables · 6 agents queued" }) except Exception as e: log.error(f"[PB SUBMIT] {e}", exc_info=True) return jsonify({"error": str(e)}), 500 @app.route("/mga/onboard", methods=["POST"]) def mga_onboard(): """ Save completed MGA onboarding to mga_onboarding table. POST /mga/onboard Body: { master_broker_code, company_name, email, license_number, state_code, city, phone, lines_of_business, verified_by } """ try: from sqlalchemy import text as sqlt data = request.get_json(force=True) or {} code = data.get("master_broker_code", "") if not code: return jsonify({"error": "master_broker_code required"}), 400 eng = get_engine() with eng.begin() as conn: conn.execute(sqlt(""" INSERT INTO mga_onboarding ( master_broker_code, company_name, email, license_number, state_code, city, phone, lines_of_business, verified_by, verification_source, is_active, onboarded_at ) VALUES ( :code, :name, :email, :license, :state, :city, :phone, :lob, :verified_by, :source, 1, NOW() ) ON DUPLICATE KEY UPDATE company_name = VALUES(company_name), license_number = VALUES(license_number), state_code = VALUES(state_code), city = VALUES(city), lines_of_business = VALUES(lines_of_business), updated_at = NOW() """), { "code": code, "name": data.get("company_name"), "email": data.get("email"), "license": data.get("license_number"), "state": data.get("state_code"), "city": data.get("city"), "phone": data.get("phone"), "lob": data.get("lines_of_business"), "verified_by": data.get("verified_by", "WEB_SEARCH"), "source": data.get("verification_source"), }) log.info(f"[MGA] Onboarded: {code} — {data.get('company_name')}") return jsonify({ "success": True, "master_broker_code": code, "message": f"MGA {data.get('company_name')} onboarded as {code}" }) except Exception as e: log.error(f"[MGA ONBOARD] {e}", exc_info=True) return jsonify({"error": str(e)}), 500 @app.route("/mga/list", methods=["GET"]) def mga_list(): """Return all onboarded MGAs.""" try: import pandas as pd eng = get_engine() with eng.connect() as conn: df = pd.read_sql( "SELECT master_broker_code, company_name, state_code, city, " "license_number, lines_of_business, onboarded_at " "FROM mga_onboarding WHERE is_active=1 ORDER BY onboarded_at DESC", conn ) records = df.where(df.notna(), other=None).to_dict(orient="records") for r in records: r["onboarded_at"] = str(r.get("onboarded_at", "")) return jsonify({"mgas": records, "count": len(records)}) except Exception as e: return jsonify({"error": str(e), "mgas": [], "count": 0}), 500 @app.route("/policybridge/submissions", methods=["GET"]) def policybridge_submissions(): """Recent PolicyBridge submissions.""" limit = request.args.get("limit", 15, type=int) try: import pandas as pd eng = get_engine() with eng.connect() as conn: df = pd.read_sql( f"""SELECT submission_id, broker_name, mga_code, insured_full_name, prop_city, prop_state, coverage_type_code, coverage_limit, pipeline_status, final_outcome, submitted_at FROM policybridge_submissions ORDER BY submitted_at DESC LIMIT {limit}""", conn ) records = df.where(df.notna(), other=None).to_dict(orient="records") for r in records: r["submitted_at"] = str(r.get("submitted_at", "")) return jsonify({"submissions": records, "count": len(records)}) except Exception as e: return jsonify({"error": str(e), "submissions": [], "count": 0}), 500 # ════════════════════════════════════════════════════════════════════ # CLAIMSENSE — CLAIMS TRIAGE PIPELINE (Agents 7–11) # ════════════════════════════════════════════════════════════════════ claim_ctx: dict = {} # in-memory pipeline context per claim def _get_or_load_claim(claim_id: str) -> dict | None: """Load claim from silver_claims if not already in claim_ctx.""" if claim_id in claim_ctx and claim_ctx[claim_id].get('claim'): return claim_ctx[claim_id]['claim'] try: import pandas as pd eng = get_engine() with eng.connect() as conn: df = pd.read_sql( f"SELECT * FROM silver_claims WHERE claim_id = %(cid)s LIMIT 1", conn, params={"cid": claim_id} ) if df.empty: return None row = df.where(df.notna(), other=None).iloc[0].to_dict() for col in ["reported_at","triaged_at","settled_at","created_at","updated_at"]: if row.get(col): row[col] = str(row[col]) # Try to parse raw_fnol_payload try: if row.get("raw_fnol_payload"): row["_fnol"] = json.loads(row["raw_fnol_payload"]) except Exception: row["_fnol"] = {} return row except Exception as e: log.error(f"[CLAIM] _get_or_load_claim({claim_id}): {e}") return None def _update_claim_status(claim_id: str, triage_result: dict): """Write final triage result back to silver_claims.""" try: from sqlalchemy import text as sqlt eng = get_engine() with eng.begin() as conn: conn.execute(sqlt(""" UPDATE silver_claims SET triage_outcome = :outcome, triage_priority = :priority, assigned_adjuster_queue = :queue, fraud_score = :fraud_score, severity_band = :severity_band, reserve_estimate = :reserve, coverage_status = :coverage_status, pipeline_status = 'TRIAGED', triaged_at = NOW(), updated_at = NOW() WHERE claim_id = :claim_id """), { "claim_id": claim_id, "outcome": triage_result.get("triage_outcome"), "priority": triage_result.get("triage_priority"), "queue": triage_result.get("adjuster_queue"), "fraud_score": triage_result.get("fraud_score"), "severity_band": triage_result.get("severity_band"), "reserve": triage_result.get("reserve_estimate"), "coverage_status": triage_result.get("coverage_status"), }) log.info(f"[CLAIM] Updated {claim_id} → TRIAGED / {triage_result.get('triage_outcome')}") except Exception as e: log.warning(f"[CLAIM] Could not update {claim_id}: {e}") def _save_claim_audit(claim_id: str, triage_result: dict): """Write AXIOM audit record to gold_claim_audit_log.""" try: from sqlalchemy import text as sqlt eng = get_engine() with eng.begin() as conn: r = conn.execute(sqlt(""" INSERT INTO gold_claim_audit_log ( claim_id, policy_number, triage_outcome, coverage_status, fraud_score, severity_band, reserve_estimate, triage_priority, overall_summary, decision_factors_json, fraud_flags_json, audit_score, audited_at ) VALUES ( :claim_id, :policy_number, :triage_outcome, :coverage_status, :fraud_score, :severity_band, :reserve_estimate, :triage_priority, :overall_summary, :factors_json, :flags_json, :audit_score, NOW() ) """), { "claim_id": claim_id, "policy_number": triage_result.get("policy_number"), "triage_outcome": triage_result.get("triage_outcome"), "coverage_status": triage_result.get("coverage_status"), "fraud_score": triage_result.get("fraud_score"), "severity_band": triage_result.get("severity_band"), "reserve_estimate":triage_result.get("reserve_estimate"), "triage_priority": triage_result.get("triage_priority"), "overall_summary": triage_result.get("overall_summary"), "factors_json": json.dumps(triage_result.get("decision_factors", [])), "flags_json": json.dumps(triage_result.get("fraud_flags", [])), "audit_score": triage_result.get("audit_score", 50), }) triage_result["audit_log_id"] = r.lastrowid log.info(f"[CLAIM] Audit saved audit_log_id={triage_result['audit_log_id']}") except Exception as e: log.warning(f"[CLAIM] Could not save audit for {claim_id}: {e}") triage_result["audit_log_id"] = None # ── CLAIM SUBMIT ───────────────────────────────────────────── @app.route("/claim/submit", methods=["POST"]) def claim_submit(): """ Submit a new FNOL claim. Supports both linked (submission_id provided) and standalone flows. POST /claim/submit Body: { claim_id, policy_number, claimant_name, incident_date, incident_type, [submission_id], [incident_description], ... } """ try: from sqlalchemy import text as sqlt data = request.get_json(force=True) if not data: return jsonify({"error": "No JSON payload"}), 400 claim_id = data.get("claim_id", "") policy_number = data.get("policy_number", "") if not claim_id: return jsonify({"error": "claim_id is required"}), 400 if not policy_number: return jsonify({"error": "policy_number is required"}), 400 eng = get_engine() with eng.begin() as conn: conn.execute(sqlt(""" INSERT INTO silver_claims ( claim_id, submission_id, policy_number, coverage_type_code, claimant_name, claimant_email, claimant_phone, claimant_relation, incident_date, incident_type, incident_description, incident_address, incident_city, incident_state, pipeline_status, raw_fnol_payload, reported_at ) VALUES ( :claim_id, :submission_id, :policy_number, :coverage_type_code, :claimant_name, :claimant_email, :claimant_phone, :claimant_relation, :incident_date, :incident_type, :incident_description, :incident_address, :incident_city, :incident_state, 'RECEIVED', :raw_payload, NOW() ) ON DUPLICATE KEY UPDATE pipeline_status = 'RECEIVED', reported_at = NOW() """), { "claim_id": claim_id, "submission_id": data.get("submission_id"), "policy_number": policy_number, "coverage_type_code": data.get("coverage_type_code"), "claimant_name": data.get("claimant_name", ""), "claimant_email": data.get("claimant_email"), "claimant_phone": data.get("claimant_phone"), "claimant_relation": data.get("claimant_relation", "Named Insured"), "incident_date": data.get("incident_date"), "incident_type": data.get("incident_type", "OTHER"), "incident_description": data.get("incident_description"), "incident_address": data.get("incident_address"), "incident_city": data.get("incident_city"), "incident_state": data.get("incident_state"), "raw_payload": json.dumps(data), }) # Queue all 5 agents for code, step in [ ("FNOL_Intake_Agent", 1), ("Coverage_Verification_Agent", 2), ("Fraud_Signal_Agent", 3), ("Severity_Reserve_Agent", 4), ("Triage_Routing_Agent", 5), ]: conn.execute(sqlt(""" INSERT IGNORE INTO silver_claim_agent_log (claim_id, agent_code, pipeline_step, status, invoked_at) VALUES (:cid, :code, :step, 'QUEUED', NOW()) """), {"cid": claim_id, "code": code, "step": step}) # Store in context claim_ctx[claim_id] = {"claim": data} log.info(f"[CLAIM] Submitted {claim_id} — policy={policy_number}") return jsonify({ "success": True, "claim_id": claim_id, "policy_number": policy_number, "message": f"{claim_id} committed to silver · 5 agents queued" }) except Exception as e: log.error(f"[CLAIM SUBMIT] {e}", exc_info=True) return jsonify({"error": str(e)}), 500 # ── AGENT 7: FNOL INTAKE ───────────────────────────────────── @app.route("/claim/agent/fnol", methods=["POST"]) def claim_agent_fnol(): claim_id = request.json.get("claim_id") if not claim_id: return jsonify({"error": "claim_id required"}), 400 claim = claim_ctx.get(claim_id, {}).get("claim") or _get_or_load_claim(claim_id) if not claim: return jsonify({"error": f"Claim {claim_id} not found"}), 404 fnol_payload = { "claim_id": claim_id, "policy_number": claim.get("policy_number", ""), "claimant_name": claim.get("claimant_name", ""), "claimant_email": claim.get("claimant_email"), "claimant_phone": claim.get("claimant_phone"), "incident_date": claim.get("incident_date") or str(claim.get("incident_date","")), "incident_type": claim.get("incident_type", "OTHER"), "incident_description": claim.get("incident_description"), "incident_address": claim.get("incident_address"), "incident_city": claim.get("incident_city"), "incident_state": claim.get("incident_state"), "estimated_damage": (claim.get("_fnol") or claim).get("estimated_damage"), "has_police_report": (claim.get("_fnol") or claim).get("has_police_report", False), "has_photos": (claim.get("_fnol") or claim).get("has_photos", False), } log.info(f"[FNOL] Running Agent 7 for {claim_id}") try: from agent7_fnol_intake import run_fnol_intake_agent result = run_fnol_intake_agent(fnol_payload) except Exception as e: log.error(f"[FNOL] Agent error: {e}") result = { "claim_id": claim_id, "status": "FNOL_ACCEPTED", "normalised_fields": fnol_payload, "completeness_score": 60, "days_to_report": 5, "late_report": False, "validation_issues": [], "incident_type": fnol_payload.get("incident_type", "OTHER"), "incident_date": str(fnol_payload.get("incident_date", "")), "policy_number": fnol_payload.get("policy_number", ""), "claimant_name": fnol_payload.get("claimant_name", ""), "_fallback": True } claim_ctx.setdefault(claim_id, {}).update({"fnol": result, "claim": claim}) return jsonify(safe_json(result)) # ── AGENT 8: COVERAGE VERIFICATION ─────────────────────────── @app.route("/claim/agent/coverage", methods=["POST"]) def claim_agent_coverage(): claim_id = request.json.get("claim_id") if not claim_id: return jsonify({"error": "claim_id required"}), 400 ctx = claim_ctx.get(claim_id, {}) fnol = ctx.get("fnol", {"claim_id": claim_id, "status": "FNOL_ACCEPTED", "incident_type": "OTHER", "normalised_fields": {}}) claim = ctx.get("claim") or _get_or_load_claim(claim_id) or {} # Try to get linked submission for policy details sub_id = claim.get("submission_id") sub = {} if sub_id: sub = load_submission(sub_id) or {} # Fallback: build minimal sub from claim data if not sub: sub = { "_coverage_type_code": claim.get("coverage_type_code", "HO-3"), "_requested_coverage_limit": 300000, "_requested_deductible": 2500, } log.info(f"[COVERAGE] Running Agent 8 for {claim_id}") try: from agent8_coverage_verify import run_coverage_verify_agent result = run_coverage_verify_agent(fnol, sub) except Exception as e: log.error(f"[COVERAGE] Agent error: {e}") result = { "claim_id": claim_id, "status": "COVERAGE_COVERED", "coverage_status": "COVERED", "coverage_type": "HO-3", "policy_number": claim.get("policy_number", ""), "applicable_limit": 300000, "deductible": 2500, "estimated_damage": 0, "net_payable_est": 0, "exclusions": [], "partial_notes": [], "summary": "Coverage verified (fallback).", "_fallback": True } claim_ctx.setdefault(claim_id, {}).update({"coverage": result}) return jsonify(safe_json(result)) # ── AGENT 9: FRAUD SIGNAL ──────────────────────────────────── @app.route("/claim/agent/fraud", methods=["POST"]) def claim_agent_fraud(): claim_id = request.json.get("claim_id") if not claim_id: return jsonify({"error": "claim_id required"}), 400 ctx = claim_ctx.get(claim_id, {}) fnol = ctx.get("fnol", {"claim_id": claim_id, "status": "FNOL_ACCEPTED", "days_to_report": 5, "normalised_fields": {}, "completeness_score": 70}) coverage = ctx.get("coverage", {"applicable_limit": 300000, "deductible": 2500}) sub = ctx.get("submission") or {} log.info(f"[FRAUD] Running Agent 9 for {claim_id}") try: from agent9_fraud_signal import run_fraud_signal_agent result = run_fraud_signal_agent(fnol, coverage, sub) except Exception as e: log.error(f"[FRAUD] Agent error: {e}") result = { "claim_id": claim_id, "status": "FRAUD_SCORED", "fraud_score": 20, "fraud_probability": 0.2, "fraud_band": "LOW", "fraud_flags": [], "_fallback": True } claim_ctx.setdefault(claim_id, {}).update({"fraud": result}) return jsonify(safe_json(result)) # ── AGENT 10: SEVERITY & RESERVE ───────────────────────────── @app.route("/claim/agent/severity", methods=["POST"]) def claim_agent_severity(): claim_id = request.json.get("claim_id") if not claim_id: return jsonify({"error": "claim_id required"}), 400 ctx = claim_ctx.get(claim_id, {}) fnol = ctx.get("fnol", {"claim_id": claim_id, "incident_type": "OTHER", "days_to_report": 5, "normalised_fields": {}}) coverage = ctx.get("coverage", {"applicable_limit": 300000, "deductible": 2500, "net_payable_est": 0}) fraud = ctx.get("fraud", {"fraud_score": 20, "fraud_band": "LOW"}) sub = ctx.get("submission") or {} log.info(f"[SEVERITY] Running Agent 10 for {claim_id}") try: from agent10_severity import run_severity_agent result = run_severity_agent(fnol, coverage, fraud, sub) except Exception as e: log.error(f"[SEVERITY] Agent error: {e}") result = { "claim_id": claim_id, "status": "SEVERITY_SCORED", "severity_band": "MODERATE", "reserve_estimate": 15000, "reserve_note": "Fallback estimate", "_fallback": True } claim_ctx.setdefault(claim_id, {}).update({"severity": result}) return jsonify(safe_json(result)) # ── AGENT 11: TRIAGE & ROUTING ──────────────────────────────── @app.route("/claim/agent/triage", methods=["POST"]) def claim_agent_triage(): claim_id = request.json.get("claim_id") if not claim_id: return jsonify({"error": "claim_id required"}), 400 ctx = claim_ctx.get(claim_id, {}) fnol = ctx.get("fnol", {"claim_id": claim_id, "status": "FNOL_ACCEPTED", "incident_type": "OTHER", "completeness_score": 70, "days_to_report": 5, "validation_issues": [], "normalised_fields": {}}) coverage = ctx.get("coverage", {"coverage_status": "COVERED", "exclusions": []}) fraud = ctx.get("fraud", {"fraud_score": 20, "fraud_band": "LOW", "fraud_flags": []}) severity = ctx.get("severity", {"severity_band": "MODERATE", "reserve_estimate": 15000, "reserve_note": ""}) log.info(f"[TRIAGE] Running Agent 11 for {claim_id}") try: from agent11_triage import run_triage_agent result = run_triage_agent(fnol, coverage, fraud, severity) except Exception as e: log.error(f"[TRIAGE] Agent error: {e}") result = { "claim_id": claim_id, "status": "TRIAGE_COMPLETE", "triage_outcome": "ADJUSTER_REVIEW", "triage_priority": "MEDIUM", "adjuster_queue": "STANDARD_ADJUSTER_QUEUE", "overall_summary": "Routed to adjuster (fallback).", "decision_factors": [], "_fallback": True } # Enrich triage result with policy number claim = ctx.get("claim") or {} result["policy_number"] = claim.get("policy_number") # Compute audit score fraud_score = int(fraud.get("fraud_score") or 0) cov_ok = 1 if coverage.get("coverage_status") == "COVERED" else 0 fnol_ok = 1 if fnol.get("status") == "FNOL_ACCEPTED" else 0 result["audit_score"] = int( (fnol_ok * 30) + (cov_ok * 30) + ((100 - fraud_score) * 0.4) ) result["fraud_flags"] = fraud.get("fraud_flags", []) # Save to DB _update_claim_status(claim_id, result) _save_claim_audit(claim_id, result) claim_ctx.pop(claim_id, None) return jsonify(safe_json(result)) # ── FULL CLAIMS PIPELINE ────────────────────────────────────── @app.route("/claim/pipeline/run", methods=["POST"]) def claim_pipeline_run(): """ Run the full 5-agent ClaimSense pipeline for a claim. POST /claim/pipeline/run Body: {"claim_id": "CLM-2026-00001"} The claim must already exist in silver_claims (via /claim/submit). """ claim_id = request.json.get("claim_id") if not claim_id: return jsonify({"error": "claim_id required"}), 400 claim = _get_or_load_claim(claim_id) if not claim: return jsonify({"error": f"Claim {claim_id} not found"}), 404 claim_ctx[claim_id] = {"claim": claim} steps = {} def call(ep): with app.test_client() as c: r = c.post(f"/claim/agent/{ep}", json={"claim_id": claim_id}, content_type="application/json") return json.loads(r.data) # Agent 7 steps["fnol"] = call("fnol") if steps["fnol"].get("status") == "FNOL_INVALID": steps["triage"] = {"triage_outcome": "DENY", "overall_summary": "FNOL invalid — missing required fields.", "triage_priority": "LOW"} return jsonify({ "final_outcome": "DENY", "halt_reason": "FNOL_INVALID", "steps": steps }) # Agent 8 steps["coverage"] = call("coverage") if steps["coverage"].get("coverage_status") == "EXCLUDED": steps["triage"] = {"triage_outcome": "DENY", "overall_summary": "Peril excluded from coverage.", "triage_priority": "LOW"} return jsonify({ "final_outcome": "DENY", "halt_reason": "COVERAGE_EXCLUDED", "steps": steps }) # Agent 9 steps["fraud"] = call("fraud") # Agent 10 steps["severity"] = call("severity") # Agent 11 steps["triage"] = call("triage") return jsonify({ "final_outcome": steps["triage"].get("triage_outcome"), "triage_priority": steps["triage"].get("triage_priority"), "adjuster_queue": steps["triage"].get("adjuster_queue"), "fraud_score": steps["fraud"].get("fraud_score"), "severity_band": steps["severity"].get("severity_band"), "reserve_estimate":steps["severity"].get("reserve_estimate"), "audit_log_id": steps["triage"].get("audit_log_id"), "overall_summary": steps["triage"].get("overall_summary"), "steps": steps, }) # ── CLAIMS READ ROUTES ──────────────────────────────────────── @app.route("/claims/recent", methods=["GET"]) def recent_claims(): limit = request.args.get("limit", 10, type=int) try: import pandas as pd eng = get_engine() with eng.connect() as conn: df = pd.read_sql( f"""SELECT claim_id, policy_number, claimant_name, incident_type, incident_date, fraud_score, severity_band, reserve_estimate, triage_outcome, triage_priority, pipeline_status, reported_at FROM silver_claims ORDER BY reported_at DESC LIMIT {limit}""", conn ) claims = [] for d in df.where(df.notna(), other=None).to_dict(orient="records"): d["reported_at"] = str(d.get("reported_at", "")) d["incident_date"]= str(d.get("incident_date", "")) claims.append(safe_json(d)) return jsonify({"claims": claims, "count": len(claims)}) except Exception as e: log.error(f"/claims/recent: {e}", exc_info=True) return jsonify({"error": str(e), "claims": [], "count": 0}), 500 @app.route("/claim/", methods=["GET"]) def get_claim(claim_id): claim = _get_or_load_claim(claim_id) if not claim: return jsonify({"error": f"Claim {claim_id} not found"}), 404 return jsonify(safe_json(claim)) @app.route("/claim/audit/", methods=["GET"]) def get_claim_audit(claim_id): try: import pandas as pd eng = get_engine() with eng.connect() as conn: df = pd.read_sql( "SELECT * FROM gold_claim_audit_log " "WHERE claim_id = %(cid)s ORDER BY audited_at DESC LIMIT 1", conn, params={"cid": claim_id} ) if df.empty: return jsonify({"error": f"No audit record for {claim_id}"}), 404 record = df.where(df.notna(), other=None).iloc[0].to_dict() for col in ["decision_factors_json", "fraud_flags_json"]: try: if record.get(col): record[col.replace("_json", "")] = json.loads(record[col]) except Exception: pass record["audited_at"] = str(record.get("audited_at", "")) return jsonify(safe_json(record)) except Exception as e: log.error(f"/claim/audit/{claim_id}: {e}", exc_info=True) return jsonify({"error": str(e)}), 500 # ── CLAIMSENSE UI ───────────────────────────────────────────── @app.route("/ui/claimsense") def claimsense(): """Serve the ClaimSense HTML frontend.""" try: html = open(os.path.join(AGENTS_DIR, "claimsense_app.html"), encoding="utf-8").read() return html, 200, {"Content-Type": "text/html; charset=utf-8"} except Exception as e: return jsonify({"error": str(e)}), 500 # ════════════════════════════════════════════════════════════════════ # RISKRADAR — GEO RISK INTELLIGENCE (Agent 12) # ════════════════════════════════════════════════════════════════════ @app.route("/risk/score", methods=["POST"]) def risk_score(): """ Score a US address for all 4 perils. POST /risk/score Body: { state_code, zip, city, [street], year_built, construction_type, roof_type, roof_year, square_footage, num_stories, [lat], [lon] } """ try: data = request.get_json(force=True) or {} if not data.get("state_code"): return jsonify({"error": "state_code is required"}), 400 from agent12_riskradar import score_address result = score_address(data) return jsonify(safe_json(result)) except Exception as e: log.error(f"[RISKRADAR] /risk/score: {e}", exc_info=True) return jsonify({"error": str(e)}), 500 @app.route("/risk/score/submission/", methods=["GET"]) def risk_score_submission(submission_id): """ Score risk for an existing submission by ID. GET /risk/score/submission/SUB-2025-00612 Pulls property data from bronze layer and runs RiskRadar. """ try: sub = load_submission(submission_id) if not sub: return jsonify({"error": f"Submission {submission_id} not found"}), 404 prop = sub.get("property") or {} insured = sub.get("insured") or {} payload = { "state_code": prop.get("state_code") or insured.get("state"), "zip": prop.get("zip") or insured.get("zip", "00000"), "city": prop.get("city") or insured.get("city", ""), "street": prop.get("street", ""), "year_built": prop.get("year_built"), "construction_type": prop.get("construction_type", "Frame"), "roof_type": prop.get("roof_type", "Asphalt Shingle"), "roof_year": prop.get("roof_year"), "square_footage": prop.get("square_footage"), "num_stories": prop.get("num_stories", 1), "lat": prop.get("latitude"), "lon": prop.get("longitude"), } from agent12_riskradar import score_address result = score_address(payload) result["submission_id"] = submission_id return jsonify(safe_json(result)) except Exception as e: log.error(f"[RISKRADAR] /risk/score/submission/{submission_id}: {e}", exc_info=True) return jsonify({"error": str(e)}), 500 @app.route("/risk/batch", methods=["POST"]) def risk_batch(): """ Score multiple addresses in one call. POST /risk/batch Body: { "addresses": [ {state_code, zip, ...}, ... ] } Max 20 addresses per request. """ try: data = request.get_json(force=True) or {} addresses = data.get("addresses", []) if not addresses: return jsonify({"error": "addresses array is required"}), 400 if len(addresses) > 20: return jsonify({"error": "Max 20 addresses per batch request"}), 400 from agent12_riskradar import score_address results = [] for addr in addresses: try: results.append(safe_json(score_address(addr))) except Exception as e: results.append({"error": str(e), "address": addr.get("street", "")}) return jsonify({"results": results, "count": len(results)}) except Exception as e: log.error(f"[RISKRADAR] /risk/batch: {e}", exc_info=True) return jsonify({"error": str(e)}), 500 @app.route("/risk/states", methods=["GET"]) def risk_states(): """Return base risk scores for all 50 states — used by UI heatmap.""" try: from agent12_riskradar import STATE_BASE_RISK, _risk_band out = {} for state, (w, fl, fi, q) in STATE_BASE_RISK.items(): composite = int(w*0.30 + fl*0.30 + fi*0.25 + q*0.15) out[state] = { "wind": w, "flood": fl, "fire": fi, "quake": q, "composite": composite, "band": _risk_band(composite) } return jsonify(out) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/ui/riskradar") def riskradar_standalone(): """Serve RiskRadar as a standalone page.""" try: html = open(os.path.join(AGENTS_DIR, "pc_insurance_multiagent_v4.html"), encoding="utf-8").read() return html, 200, {"Content-Type": "text/html; charset=utf-8"} except Exception as e: return jsonify({"error": str(e)}), 500 # ════════════════════════════════════════════════════════════════════ # ACTUARIAL OS — Routes for Agents 13–16 # ════════════════════════════════════════════════════════════════════ def _import_agent(module_name, func_name): try: mod = __import__(module_name) return getattr(mod, func_name) except Exception as e: logging.warning(f"Could not import {module_name}.{func_name}: {e}") return None @app.route("/actuarial/etl/all", methods=["POST"]) def actuarial_etl_all(): try: results = {} for module, func, key in [ ('agent13_loss_ratio', 'run_etl_bronze_to_silver', 'agent13_lr'), ('agent14_ibnr', 'run_etl_triangles_to_silver', 'agent14_ibnr'), ('agent15_cat', 'run_etl_cat_to_silver', 'agent15_cat'), ('agent16_rein_opt', 'run_etl_rein_to_silver', 'agent16_rein'), ]: try: fn = _import_agent(module, func) results[key] = fn() if fn else {'error': f'{module} not available'} except Exception as e: results[key] = {'error': str(e)} return jsonify({'status': 'ok', 'results': safe_json(results)}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route("/actuarial/loss-ratio/score", methods=["POST"]) def actuarial_lr_score(): try: from agent13_loss_ratio import run_agent13 data = request.get_json(force=True) or {} if not data.get('lob') or not data.get('state_code'): return jsonify({'error': 'lob and state_code are required'}), 400 return jsonify(safe_json(run_agent13(data))) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route("/actuarial/loss-ratio/history", methods=["GET"]) def actuarial_lr_history(): try: import pandas as pd lob = request.args.get('lob') state = request.args.get('state') limit = request.args.get('limit', 50, type=int) eng = get_engine() query = "SELECT * FROM gold_act_loss_ratio_pred WHERE 1=1" params = {} if lob: query += " AND lob=:lob"; params['lob'] = lob if state: query += " AND state_code=:state"; params['state'] = state query += f" ORDER BY predicted_at DESC LIMIT {limit}" with eng.connect() as conn: df = pd.read_sql(query, conn, params=params if params else None) records = df.where(df.notna(), other=None).to_dict(orient='records') for r in records: r['predicted_at'] = str(r.get('predicted_at','')) return jsonify({'predictions': records, 'count': len(records)}) except Exception as e: return jsonify({'error': str(e), 'predictions': []}), 500 @app.route("/actuarial/ibnr/estimate", methods=["POST"]) def actuarial_ibnr_estimate(): try: from agent14_ibnr import run_agent14 data = request.get_json(force=True) or {} if not data.get('lob') or not data.get('state_code'): return jsonify({'error': 'lob and state_code are required'}), 400 return jsonify(safe_json(run_agent14(data))) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route("/actuarial/ibnr/history", methods=["GET"]) def actuarial_ibnr_history(): try: import pandas as pd lob = request.args.get('lob') limit = request.args.get('limit', 50, type=int) eng = get_engine() query = "SELECT * FROM gold_act_ibnr_estimates WHERE 1=1" params = {} if lob: query += " AND lob=:lob"; params['lob'] = lob query += f" ORDER BY estimated_at DESC LIMIT {limit}" with eng.connect() as conn: df = pd.read_sql(query, conn, params=params if params else None) records = df.where(df.notna(), other=None).to_dict(orient='records') for r in records: r['estimated_at'] = str(r.get('estimated_at','')) return jsonify({'estimates': records, 'count': len(records)}) except Exception as e: return jsonify({'error': str(e), 'estimates': []}), 500 @app.route("/actuarial/cat/model", methods=["POST"]) def actuarial_cat_model(): try: from agent15_cat import run_agent15 data = request.get_json(force=True) or {} if not data.get('lob') or not data.get('peril'): return jsonify({'error': 'lob and peril are required'}), 400 return jsonify(safe_json(run_agent15(data))) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route("/actuarial/cat/history", methods=["GET"]) def actuarial_cat_history(): try: import pandas as pd lob = request.args.get('lob') scenario = request.args.get('scenario', 'BASE') limit = request.args.get('limit', 30, type=int) eng = get_engine() query = "SELECT * FROM gold_act_cat_model WHERE model_scenario=:scenario" params = {'scenario': scenario} if lob: query += " AND lob=:lob"; params['lob'] = lob query += f" ORDER BY modelled_at DESC LIMIT {limit}" with eng.connect() as conn: df = pd.read_sql(query, conn, params=params) records = df.where(df.notna(), other=None).to_dict(orient='records') for r in records: r['modelled_at'] = str(r.get('modelled_at','')) return jsonify({'models': records, 'count': len(records)}) except Exception as e: return jsonify({'error': str(e), 'models': []}), 500 @app.route("/actuarial/reinsurance/optimise", methods=["POST"]) def actuarial_rein_optimise(): try: from agent16_rein_opt import run_agent16 data = request.get_json(force=True) or {} if not data.get('lob'): return jsonify({'error': 'lob is required'}), 400 return jsonify(safe_json(run_agent16(data))) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route("/actuarial/reinsurance/history", methods=["GET"]) def actuarial_rein_history(): try: import pandas as pd lob = request.args.get('lob') limit = request.args.get('limit', 40, type=int) eng = get_engine() query = "SELECT * FROM gold_act_rein_optimisation WHERE 1=1" params = {} if lob: query += " AND lob=:lob"; params['lob'] = lob query += f" ORDER BY optimised_at DESC LIMIT {limit}" with eng.connect() as conn: df = pd.read_sql(query, conn, params=params if params else None) records = df.where(df.notna(), other=None).to_dict(orient='records') for r in records: r['optimised_at'] = str(r.get('optimised_at','')) return jsonify({'optimisations': records, 'count': len(records)}) except Exception as e: return jsonify({'error': str(e), 'optimisations': []}), 500 @app.route("/actuarial/portfolio/summary", methods=["GET"]) def actuarial_portfolio_summary(): try: import pandas as pd eng = get_engine() summary = {} try: with eng.connect() as conn: df = pd.read_sql( "SELECT lob, AVG(loss_ratio) AS avg_lr, SUM(earned_premium) AS total_ep " "FROM silver_act_loss_ratios GROUP BY lob", conn) for _, row in df.iterrows(): summary[row['lob']] = { 'lob': row['lob'], 'total_ep': round(float(row['total_ep'] or 0), 2), 'avg_lr': round(float(row['avg_lr'] or 0), 5), } except Exception as e: log.warning(f"[ACTUARIAL] portfolio summary: {e}") return jsonify({'by_lob': list(summary.values()), 'generated_at': str(datetime.now())}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route("/actuarial/audit", methods=["GET"]) def actuarial_audit(): try: import pandas as pd agent = request.args.get('agent') lob = request.args.get('lob') limit = request.args.get('limit', 100, type=int) eng = get_engine() query = "SELECT * FROM gold_act_audit_log WHERE 1=1" params = {} if agent: query += " AND agent_name=:agent"; params['agent'] = agent if lob: query += " AND lob=:lob"; params['lob'] = lob query += f" ORDER BY audited_at DESC LIMIT {limit}" with eng.connect() as conn: df = pd.read_sql(query, conn, params=params if params else None) records = df.where(df.notna(), other=None).to_dict(orient='records') for r in records: r['audited_at'] = str(r.get('audited_at','')) return jsonify({'audit_records': records, 'count': len(records)}) except Exception as e: return jsonify({'error': str(e), 'audit_records': []}), 500 @app.route("/actuarial/scenario/run", methods=["POST"]) def actuarial_scenario_run(): try: import uuid as _uuid from sqlalchemy import text as sqlt data = request.get_json(force=True) or {} run_id = data.get('run_id') or f"SCN-{_uuid.uuid4().hex[:12].upper()}" results = {} for lob in data.get('lob_scope', ['HO', 'AUTO']): lob_results = {} for module, func, key in [ ('agent13_loss_ratio', 'run_agent13', 'loss_ratio'), ('agent15_cat', 'run_agent15', 'cat'), ('agent16_rein_opt', 'run_agent16', 'reinsurance'), ]: try: fn = _import_agent(module, func) lob_results[key] = safe_json(fn({'lob': lob, 'run_id': run_id})) if fn else {'error': 'not available'} except Exception as e: lob_results[key] = {'error': str(e)} results[lob] = lob_results return jsonify(safe_json({'run_id': run_id, 'results': results})) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route("/actuarial/scenario/history", methods=["GET"]) def actuarial_scenario_history(): try: import pandas as pd limit = request.args.get('limit', 20, type=int) eng = get_engine() with eng.connect() as conn: df = pd.read_sql( f"SELECT * FROM gold_act_scenario_runs ORDER BY created_at DESC LIMIT {limit}", conn) records = df.where(df.notna(), other=None).to_dict(orient='records') for r in records: r['created_at'] = str(r.get('created_at','')) return jsonify({'scenarios': records, 'count': len(records)}) except Exception as e: return jsonify({'error': str(e), 'scenarios': []}), 500 @app.route("/actuarial/rate-adequacy", methods=["GET"]) def actuarial_rate_adequacy(): try: import pandas as pd lob = request.args.get('lob') state = request.args.get('state') eng = get_engine() query = "SELECT * FROM silver_act_rate_adequacy WHERE 1=1" params = {} if lob: query += " AND lob=:lob"; params['lob'] = lob if state: query += " AND state_code=:state"; params['state'] = state query += " ORDER BY analysis_date DESC LIMIT 200" with eng.connect() as conn: df = pd.read_sql(query, conn, params=params if params else None) records = df.where(df.notna(), other=None).to_dict(orient='records') for r in records: r['analysis_date'] = str(r.get('analysis_date','')) return jsonify({'adequacy': records, 'count': len(records)}) except Exception as e: return jsonify({'error': str(e), 'adequacy': []}), 500 # ════════════════════════════════════════════════════════════════════ # QUOTE COMPARISON — AI-powered market quote engine # ════════════════════════════════════════════════════════════════════ @app.route("/quote/compare", methods=["POST"]) def quote_compare(): """ Generate competitive quote comparison for a submission. Uses our own pricing agent + market benchmarks to produce 3-5 quote options with recommendation. POST /quote/compare Body: { submission_id OR full payload with property/coverage details } """ try: data = request.get_json(force=True) or {} sub_id = data.get("submission_id") payload = data # Load from DB if submission_id provided if sub_id and not data.get("property"): sub = load_submission(sub_id) if sub: payload = sub prop = payload.get("property", {}) policy = payload.get("policy_request", {}) or payload.get("policy", {}) insured= payload.get("insured", {}) cov_type = (policy.get("coverage_type") or payload.get("coverage_type_code") or "HO-3") limit = float(policy.get("requested_coverage_limit") or payload.get("coverage_limit") or 300000) ded = float(policy.get("requested_deductible") or payload.get("deductible") or 2500) state = (prop.get("state_code") or prop.get("state") or insured.get("state") or "TX") year_built = int(prop.get("year_built") or 2000) sqft = int(prop.get("square_footage") or 2000) credit = int(insured.get("credit_score") or 680) roof_year = int(prop.get("roof_year") or year_built + 5) construction = prop.get("construction_type", "Frame") # ── Run our own pricing agent for base premium ───────────── base_premium = None try: from agent4_pricing import run_pricing_agent uw_mock = {"status": "UW_APPROVED", "coverage_type": cov_type, "uw_approval_probability": 0.85, "expected_loss_ratio": 0.58, "reinsurance_required": limit > 2_000_000} prop_mock= {"status": "RISK_ACCEPTABLE", "risk_band": "MEDIUM", "peril_scores": {"wind_score": 35, "flood_score": 28, "fire_score": 22, "overall_risk": 38}} pricing = run_pricing_agent(uw_mock, prop_mock, payload) base_premium = float(pricing.get("final_premium") or pricing.get("annual_premium") or 0) except Exception as pe: log.warning(f"[QUOTE] Pricing agent error: {pe}") # Fallback: actuarial estimate age_years = max(0, 2026 - year_built) age_factor = min(age_years * 0.004, 0.25) state_mult = {"FL":1.85,"TX":1.42,"CA":1.38,"LA":1.72,"NY":1.28, "SC":1.35,"NC":1.30,"AL":1.28,"MS":1.32,"HI":1.55}.get(state, 1.05) credit_disc = max(0.88, min(1.18, 1.6 - credit / 900)) roof_age = max(0, 2026 - roof_year) roof_factor = 1 + min(roof_age * 0.012, 0.28) const_disc = 0.90 if "masonry" in construction.lower() else 1.0 # Base rate: ~0.9% of limit for standard property rate = 0.009 * state_mult * (1 + age_factor) * credit_disc * roof_factor * const_disc base_premium = round(limit * rate) if base_premium < 800: base_premium = round(limit * 0.009) import random, math rng = random.Random(hash(f"{sub_id or state}{cov_type}{limit}") % (2**31)) # ── Generate 5 competitive quotes ────────────────────────── carriers = [ {"name": "PolicySense Direct", "tier": "our_quote", "am_best": "A+", "market_share": "PolicySense Platform", "color": "#818cf8"}, {"name": "Nationwide", "tier": "tier1", "am_best": "A+", "market_share": "7.2% market share", "color": "#00d4ff"}, {"name": "State Farm", "tier": "tier1", "am_best": "A++", "market_share": "16.1% — largest US P&C", "color": "#00ff9d"}, {"name": "Progressive", "tier": "tier1", "am_best": "A+", "market_share": "6.4% market share", "color": "#f59e0b"}, {"name": "Allstate", "tier": "tier2", "am_best": "A+", "market_share": "5.5% market share", "color": "#ff6b35"}, {"name": "Travelers", "tier": "tier2", "am_best": "A++", "market_share": "3.9% market share", "color": "#a855f7"}, ] quotes = [] # Our quote — best value positioning our_annual = round(base_premium * rng.uniform(0.88, 0.95)) quotes.append({ "carrier": "PolicySense Direct", "tier": "our_quote", "am_best": "A+", "market_info": "PolicySense Platform — AI-optimised pricing", "color": "#818cf8", "annual": our_annual, "monthly": round(our_annual / 12), "deductible": int(ded), "coverage_limit": int(limit), "coverage_type": cov_type, "features": ["AI-powered claim triage", "Real-time risk monitoring", "24/7 digital portal", "AXIOM audit trail"], "pros": ["Best value for risk profile", "Instant digital policy", "No broker fees"], "cons": ["Newer carrier — less brand recognition"], "recommended": True, "savings_vs_market": None, # filled below "score": 92, }) # Market quotes — realistic range around base multipliers = [ (1.08, 0.96, ["Established brand", "Wide agent network"], ["Higher premium", "Traditional process"], 78), (1.12, 0.94, ["Largest US insurer", "Strong claim record"], ["Premium pricing", "Less flexible"], 74), (1.05, 0.97, ["Usage-based options", "Snapshot discount"], ["Variable pricing", "Telematics required"], 81), (1.18, 0.91, ["Bundle discounts", "Local agents"], ["Highest premium", "Slower claims"], 68), (1.15, 0.93, ["Strong commercial presence", "A++ rated"], ["Complex policy terms", "High deductibles"], 71), ] for i, (c, mult) in enumerate(zip(carriers[1:], multipliers)): m, disc, pros, cons, score = mult annual = round(base_premium * m * rng.uniform(0.97, 1.03)) quotes.append({ "carrier": c["name"], "tier": c["tier"], "am_best": c["am_best"], "market_info": c["market_share"], "color": c["color"], "annual": annual, "monthly": round(annual / 12), "deductible": int(ded), "coverage_limit":int(limit), "coverage_type": cov_type, "features": ["Standard homeowners coverage", "Claims helpline", f"AM Best {c['am_best']} rated"], "pros": pros, "cons": cons, "recommended": False, "savings_vs_market": None, "score": score, }) # Sort by annual premium quotes.sort(key=lambda q: q["annual"]) # Calculate savings vs market average market_avg = round(sum(q["annual"] for q in quotes[1:]) / max(len(quotes)-1, 1)) for q in quotes: saving = market_avg - q["annual"] q["savings_vs_market"] = saving q["savings_pct"] = round(saving / market_avg * 100, 1) if market_avg > 0 else 0 # AI recommendation narrative best = quotes[0] saved = market_avg - best["annual"] narrative = ( f"Based on the property profile ({state}, {cov_type}, " f"${int(limit/1000)}K limit, {year_built} build), " f"PolicySense Direct offers the strongest value at " f"${best['annual']:,}/year — saving approximately " f"${max(0,saved):,} vs the market average of ${market_avg:,}. " f"The property's {construction} construction and " f"{'newer' if 2026-roof_year<10 else 'aging'} roof " f"({'good discount applied' if 2026-roof_year<10 else 'surcharge applied'}) " f"are the primary pricing drivers." ) return jsonify(safe_json({ "submission_id": sub_id, "coverage_type": cov_type, "coverage_limit": int(limit), "deductible": int(ded), "state": state, "market_avg": market_avg, "our_premium": our_annual, "savings": max(0, market_avg - our_annual), "quotes": quotes, "recommendation": narrative, "generated_at": datetime.utcnow().isoformat(), })) except Exception as e: log.error(f"[QUOTE] /quote/compare: {e}", exc_info=True) return jsonify({"error": str(e)}), 500 # ════════════════════════════════════════════════════════════════════ # ENTRYPOINT # ════════════════════════════════════════════════════════════════════ if __name__ == "__main__": env = "HuggingFace → Clever Cloud" if _is_huggingface() else "Local → MySQL" port = int(os.environ.get("PORT", 7860 if _is_huggingface() else 5000)) debug = not _is_huggingface() log.info("=" * 60) log.info(" PolicyBridge Flask API Server") log.info(f" Environment : {env}") log.info(f" DB Host : {DB['host']}:{DB['port']}") log.info(f" DB Name : {DB['database']}") log.info(f" Port : {port}") log.info(f" Agents dir : {AGENTS_DIR}") log.info("=" * 60) for mf in ["agent1_kyc_classifier.pkl","agent2_property_risk.pkl", "agent3_underwriting.pkl","agent4_pricing.pkl"]: p = os.path.join(AGENTS_DIR,"models",mf) log.info(f" {'✓' if os.path.exists(p) else '✗ MISSING'} models/{mf}") log.info(f" ✓ agent6_audit (AXIOM) — built-in") for cf in ["agent12_riskradar.py"]: cp = os.path.join(AGENTS_DIR, cf) log.info(f" {'✓' if os.path.exists(cp) else '✗ MISSING'} {cf}") for mf in ["agent12_riskradar.pkl"]: p = os.path.join(AGENTS_DIR,"models",mf) log.info(f" {'✓' if os.path.exists(p) else '○ untrained'} models/{mf}") for cf in ["agent7_fnol_intake.py","agent8_coverage_verify.py", "agent9_fraud_signal.py","agent10_severity.py","agent11_triage.py", "agent12_riskradar.py"]: cp = os.path.join(AGENTS_DIR, cf) log.info(f" {'✓' if os.path.exists(cp) else '✗ MISSING'} {cf}") for mf in ["agent9_fraud_signal.pkl","agent10_severity.pkl"]: p = os.path.join(AGENTS_DIR,"models",mf) log.info(f" {'✓' if os.path.exists(p) else '○ untrained'} models/{mf}") log.info("=" * 60) app.run(host="0.0.0.0", port=port, debug=debug, use_reloader=False)