Fraud Detection & Risk Scoring: Architecture Blueprint & Production Stack

from fastapi import FastAPI
from pydantic import BaseModel
from neo4j import GraphDatabase
import xgboost as xgb
import numpy as np
import time

app = FastAPI(title="Real-Time Fraud Risk Scoring Engine")

# Neo4j Driver Connection
neo4j_driver = GraphDatabase.driver(
    "bolt://neo4j-cluster.internal:7687",
    auth=("neo4j", "VPC_SECURE_PASSWORD")
)

# Load Pre-trained XGBoost Risk Model
risk_model = xgb.Booster()
risk_model.load_model("/models/fraud_xgboost_v4.json")

class TransactionPayload(BaseModel):
    user_id: str
    device_hash: str
    amount_usd: float
    ip_address: str

class RiskDecision(BaseModel):
    action: str  # ALLOW, STEP_UP_MFA, BLOCK
    risk_score: float
    latency_ms: float

def query_shared_entity_count(tx: any, user_id: str, device_hash: str) -> int:
    """Cypher query counting distinct accounts sharing the same device fingerprint."""
    query = """
    MATCH (u:User {id: $user_id})-[:USED_DEVICE]->(d:Device {hash: $device_hash})<-[:USED_DEVICE]-(other:User)
    RETURN count(distinct other) AS shared_user_count
    """
    result = tx.run(query, user_id=user_id, device_hash=device_hash)
    record = result.single()
    return record["shared_user_count"] if record else 0

@app.post("/v1/evaluate_transaction", response_model=RiskDecision)
async def evaluate_fraud_risk(payload: TransactionPayload):
    start_time = time.perf_counter()
    
    # 1. Execute Sub-15ms Neo4j Subgraph Query
    with neo4j_driver.session() as session:
        shared_accounts = session.execute_read(
            query_shared_entity_count, payload.user_id, payload.device_hash
        )
    
    # 2. Assemble Feature Vector
    features = np.array([[payload.amount_usd, shared_accounts, 1.0 if payload.amount_usd > 2500 else 0.0]])
    dmatrix = xgb.DMatrix(features)
    
    # 3. Model Inference
    risk_score = float(risk_model.predict(dmatrix)[0])
    latency_ms = (time.perf_counter() - start_time) * 1000.0
    
    # 4. Decision Threshold Routing
    if risk_score > 0.85:
        action = "BLOCK"
    elif risk_score > 0.65:
        action = "STEP_UP_MFA"
    else:
        action = "ALLOW"
        
    return RiskDecision(
        action=action,
        risk_score=round(risk_score, 4),
        latency_ms=round(latency_ms, 2)
    )