AI-powered fraud detection in insurance: a hands-on implementation guide for claims teams AI-powered fraud detection in insurance: a hands-on implementation guide for claims teams

A top-20 U.S. P&C insurer cut suspicious-claim leakage by 15% in eight months by moving from a rules-only engine to a real-time AI pipeline. The project required re-engineering the data ingestion layer, retraining legacy models, and training adjusters to trust the model’s “medium-risk” verdicts. This guide details the implementation steps, code, and cost/benefit numbers.

By the end of this guide, you will have:

A streaming pipeline that ingests FNOL, adjuster notes, repair invoices, and third-party data (LexisNexis, MVR, CLUE) in under 50 ms. A production-ready anomaly model (XGBoost + SHAP) that flags claims with a precision ≥ 70% and recall ≥ 60%.

An alerting dashboard that allows adjusters to override without engineering tickets. A rollback plan that reverts to the old rules engine if the model drifts > 8% on validation data.

• Estimated build time: 6–8 weeks for a squad of 1 FTE data engineer, 1 FTE ML engineer, 1 part-time claims analyst, and 20 hours of vendor support. Budget: $95k (AWS, Snowflake, Fivetran, open-source models). Step 1 – Instrument the data layer
• Insurance fraud detection begins with data that adjusters already collect but rarely share in real time. 1.1 Add three new CDC streams
• Adjusters type most of the signal into legacy case management systems. To capture this without touching the UI, we use database change-data-capture (CDC). Data source
• CDC tool Table pattern

Latency target Cost/month (AWS)

Guidewire ClaimCenter Debezium MySQL connector

claim, activity, diarization 300 ms

$1,200 Repair invoice PDFs

Amazon Textract + S3 event invoice_text

Textract OCR 0.22

Adjuster notes churn_words_score

CREATE OR REPLACE TABLE CLAIM_ENVELOPE (
    claim_key       STRING,
    policy_id       STRING,
    loss_date       TIMESTAMP_LTZ,
    claim_status    STRING,
    adjuster_notes  VARIANT,
    repair_invoices VARIANT,
    risk_report     VARIANT,
    updated_at      TIMESTAMP_LTZ,
    valid_from      TIMESTAMP_LTZ,
    valid_to        TIMESTAMP_LTZ,
    is_current      BOOLEAN
);

NLP pipeline 0.15

-- Debezium → Kafka → Snowpipe → CLAIM_ENVELOPE
COPY INTO CLAIM_ENVELOPE
FROM @KAFKA_CLAIMS_STAGE/file_01.json
FILE_FORMAT = (TYPE = 'JSON' STRIP_OUTER_ARRAY = TRUE);

Third-party risk_score

LexisNexis 0.25

Network repair_shop_distance_km

CREATE OR REPLACE TASK DATA_FRESHNESS_WATCHDOG
WAREHOUSE = COMPUTE_WH
SCHEDULE = '1 MINUTE'
AS
INSERT INTO FRESHNESS_ALERTS
SELECT
    CURRENT_TIMESTAMP() as alert_time,
    'CLAIM_ENVELOPE' as table_name,
    DATEDIFF(second, MAX(updated_at), CURRENT_TIMESTAMP()) as lag_seconds
FROM CLAIM_ENVELOPE
QUALIFY 1=1
HAVING lag_seconds > 120; -- 2-minute SLA

Google Maps API 0.20

3.3 Model training script (XGBoost) 3.4 Validation metrics

Precision@5%: 72% Recall@5%: 61%

F1@5%: 0.66 SHAP feature importance matched adjuster intuition: repair_burst and risk_score topped the list.

from feast import Entity, ValueType

claim_entity = Entity(
    name="claim_key",
    value_type=ValueType.STRING,
    description="Unique identifier for an insurance claim"
)

Step 4 – Serve the model in production 4.1 Batch scoring (daily SIU queue)

• Every midnight, pull claims updated in the last 24 hours and append the model score. 4.2 Real-time API (REST)
• Adjusters see the score within 80 ms via a Lambda function fronted by API Gateway. Step 5 – Build the adjuster alerting layer
• Adjusters refused to trust a black-box number. We built a simple three-tier UI in Looker embedded in Guidewire. Tier
• Score range Adjuster action

from feast import FeatureView, Field
from feast.types import Float32

velocity_fv = FeatureView(
    name="claim_velocity_365d",
    entities=["claim_key"],
    schema=[
        Field(name="velocity_365d", dtype=Float32)
    ],
    source=SnowflakeSource(
        table="FEATURE_VELOCITY",
        event_timestamp_column="event_ts"
    ),
    ttl=timedelta(days=365)
)

Override reason codes Low

0.0–0.25 Auto-close

import redis, json

r = redis.Redis(host="redis-fs.example.com", port=6379)

feature_vector = {
    "velocity_365d": 5.2,
    "repair_burst": 3.7,
    "adjuster_churn_words": 0.84,
    "risk_score": 0.61
}

r.hset("claim:CLAIM-2024-12345", mapping=feature_vector)

— Medium

0.26–0.60 Manual review

12 codes (e.g., "legit repair shop", "medical records pending") High

0.61–1.00 SIU referral

4 codes (e.g., "organized ring suspected", "prior conviction") The override rate after 90 days was 18% for medium-risk claims, confirming adjusters still trusted their judgment. The override flag became a new feature for model retraining.

Step 6 – Monitor and retrain 6.1 Drift detection

import xgboost as xgb
from sklearn.model_selection import train_test_split
from sklearn.metrics import precision_recall_curve

params = {
    "objective": "binary:logistic",
    "eval_metric": "aucpr",
    "max_depth": 6,
    "learning_rate": 0.05,
    "subsample": 0.8,
    "colsample_bytree": 0.8,
    "scale_pos_weight": 1 / 0.08
}

X_train, X_val, y_train, y_val = train_test_split(
    features, labels, test_size=0.2, random_state=42
)

dtrain = xgb.DMatrix(X_train, label=y_train)
dval = xgb.DMatrix(X_val, label=y_val)

model = xgb.train(
    params,
    dtrain,
    num_boost_round=1000,
    early_stopping_rounds=20,
    evals=[(dval, "val")]
)

# Save model artifact to S3
model.save_model("s3://ai-fraud-model/xgboost_fraud_20240601.model")

-- Snowflake stored procedure
CREATE OR REPLACE PROCEDURE SCORE_NEW_CLAIMS()
RETURNS STRING
LANGUAGE PYTHON
AS
$$
import xgboost as xgb
import pandas as pd
import joblib

model = joblib.load('/tmp/xgboost_fraud_20240601.model')
cursor = snowflake.snowpark_session.cursor()

df = cursor.execute("""
    SELECT claim_key, velocity_365d, repair_burst,
           churn_words_score, risk_score, repair_shop_distance_km
    FROM CLAIM_ENVELOPE
    WHERE updated_at >= DATEADD(day, -1, CURRENT_DATE())
""").to_pandas()

df['fraud_score'] = model.predict_proba(df[features])[:, 1]

for _, row in df.iterrows():
    cursor.execute(f"""
        INSERT INTO FRAUD_SCORES
        VALUES ('{row['claim_key']}', {row['fraud_score']}, CURRENT_TIMESTAMP())
    """)
$$;

import boto3
import joblib
import numpy as np

model = joblib.load('s3://ai-fraud-model/xgboost_fraud_20240601.model')

def lambda_handler(event, context):
    claim_key = event['queryStringParameters']['claim_key']
    features = retrieve_features_from_redis(claim_key)
    score = model.predict_proba(np.array(features).reshape(1, -1))[0][1]
    return {
        'statusCode': 200,
        'body': json.dumps({'claim_key': claim_key, 'fraud_score': float(score)})
    }

from alibi_detect import KSDrift

drift_detector = KSDrift(
    p_val=0.05,
    preprocess_fn=preprocessor,
    X_ref=reference_data
)

preds = drift_detector.predict(new_data)
if preds['data']['is_drift']:
    send_email("drift@claims.example.com", "Feature drift detected")