PCAgentinAI / app.py
ITNovaML's picture
Fix: policybridge_submit writes real insured/property IDs to bronze + fix premium calc
05ff1bf
Raw
History Blame Contribute Delete
124 kB
"""
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/<sub_id>", methods=["GET"])
def get_audit(sub_id):
"""
GET /audit/<submission_id>
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/<sub_id>", methods=["POST"])
def trigger_audit(sub_id):
"""
POST /audit/run/<submission_id>
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/<claim_id>", 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/<claim_id>", 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/<submission_id>", 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)