Spaces:
Sleeping
Sleeping
File size: 6,089 Bytes
9d20d0b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 |
"""Fraud Risk Agent - Model Contract Implementation
This module implements the fraud-risk-agent model with strict JSON contract.
Decision output: investigate | allow
"""
import json
from typing import Dict, List, Any
class FraudRiskAgent:
"""Fraud Risk Decision Agent with formal model contract."""
def __init__(self):
self.model_version = "1.0.0"
self.decision_threshold = 0.65
def analyze(self, claim_data: Dict[str, Any]) -> Dict[str, Any]:
"""Analyze claim and return decision contract.
Args:
claim_data: Structured claim information
Returns:
Model contract (STRICT JSON):
{
"fraud_score": float,
"risk_band": "low | medium | high",
"top_indicators": list,
"recommended_action": "investigate | allow",
"confidence": float,
"explainability": {
"signals": list,
"weights": dict
}
}
"""
# Extract features
amount = claim_data.get('amount', 0)
claim_type = claim_data.get('type', 'unknown')
claimant_history = claim_data.get('claimant_history', {})
# Calculate fraud indicators
indicators = self._calculate_indicators(claim_data)
fraud_score = self._calculate_fraud_score(indicators)
risk_band = self._determine_risk_band(fraud_score)
# Determine action
recommended_action = "investigate" if fraud_score >= self.decision_threshold else "allow"
# Build explainability
explainability = self._build_explainability(indicators)
# Return strict model contract
return {
"fraud_score": round(fraud_score, 3),
"risk_band": risk_band,
"top_indicators": self._get_top_indicators(indicators, n=5),
"recommended_action": recommended_action,
"confidence": round(self._calculate_confidence(indicators), 3),
"explainability": explainability
}
def _calculate_indicators(self, claim_data: Dict[str, Any]) -> Dict[str, float]:
"""Calculate fraud indicators from claim data."""
indicators = {}
# Amount deviation
amount = claim_data.get('amount', 0)
avg_amount = claim_data.get('average_claim_amount', 5000)
indicators['amount_deviation'] = abs(amount - avg_amount) / avg_amount if avg_amount > 0 else 0
# Frequency signal
claim_count = claim_data.get('claimant_history', {}).get('claim_count', 0)
indicators['high_frequency'] = min(claim_count / 10.0, 1.0)
# Temporal pattern
days_since_policy = claim_data.get('days_since_policy_start', 365)
indicators['early_claim'] = 1.0 if days_since_policy < 30 else 0.0
# Document consistency
doc_score = claim_data.get('document_consistency_score', 1.0)
indicators['document_mismatch'] = 1.0 - doc_score
# Entity linkage
linked_entities = claim_data.get('linked_suspicious_entities', 0)
indicators['entity_linkage'] = min(linked_entities / 5.0, 1.0)
return indicators
def _calculate_fraud_score(self, indicators: Dict[str, float]) -> float:
"""Calculate weighted fraud score."""
weights = {
'amount_deviation': 0.25,
'high_frequency': 0.20,
'early_claim': 0.15,
'document_mismatch': 0.25,
'entity_linkage': 0.15
}
score = sum(indicators.get(k, 0) * w for k, w in weights.items())
return min(max(score, 0.0), 1.0)
def _determine_risk_band(self, fraud_score: float) -> str:
"""Determine risk band from fraud score."""
if fraud_score >= 0.7:
return "high"
elif fraud_score >= 0.4:
return "medium"
else:
return "low"
def _calculate_confidence(self, indicators: Dict[str, float]) -> float:
"""Calculate confidence in the decision."""
# Higher confidence when indicators are consistent
variance = sum((v - 0.5) ** 2 for v in indicators.values()) / len(indicators)
confidence = 1.0 - (variance * 2)
return min(max(confidence, 0.0), 1.0)
def _get_top_indicators(self, indicators: Dict[str, float], n: int = 5) -> List[str]:
"""Get top N fraud indicators."""
sorted_indicators = sorted(indicators.items(), key=lambda x: x[1], reverse=True)
return [k for k, v in sorted_indicators[:n] if v > 0.1]
def _build_explainability(self, indicators: Dict[str, float]) -> Dict[str, Any]:
"""Build explainability payload."""
signals = []
for indicator, value in indicators.items():
if value > 0.1:
signals.append({
"indicator": indicator,
"value": round(value, 3),
"description": self._get_indicator_description(indicator)
})
weights = {
'amount_deviation': 0.25,
'high_frequency': 0.20,
'early_claim': 0.15,
'document_mismatch': 0.25,
'entity_linkage': 0.15
}
return {
"signals": signals,
"weights": weights
}
def _get_indicator_description(self, indicator: str) -> str:
"""Get human-readable description of indicator."""
descriptions = {
'amount_deviation': 'Claim amount significantly differs from average',
'high_frequency': 'Claimant has high claim frequency',
'early_claim': 'Claim filed shortly after policy inception',
'document_mismatch': 'Inconsistencies detected in documentation',
'entity_linkage': 'Claimant linked to suspicious entities'
}
return descriptions.get(indicator, indicator)
|