| """Pattern Analysis Agent - Detects known fraud patterns.""" |
|
|
| from typing import Dict, List, Any |
|
|
|
|
| class PatternAnalysisAgent: |
| """Analyzes claims for known fraud patterns.""" |
| |
| def __init__(self): |
| self.name = "PatternAnalysisAgent" |
| self.version = "1.0" |
| self.known_patterns = [ |
| "rapid_succession_claims", |
| "round_number_amounts", |
| "weekend_incidents", |
| "similar_claim_history" |
| ] |
| |
| def process(self, claim_data: Dict[str, Any], historical_data: Dict[str, Any]) -> Dict[str, Any]: |
| """Detect known fraud patterns.""" |
| detected_patterns = [] |
| pattern_scores = {} |
| |
| |
| if historical_data.get("prior_claims", 0) >= 3: |
| detected_patterns.append("rapid_succession_claims") |
| pattern_scores["rapid_succession"] = 0.7 |
| |
| |
| amount = claim_data.get("claim_amount", 0) |
| if amount % 1000 == 0 and amount > 0: |
| detected_patterns.append("round_number_amounts") |
| pattern_scores["round_numbers"] = 0.5 |
| |
| |
| overall_score = sum(pattern_scores.values()) / len(self.known_patterns) if pattern_scores else 0.0 |
| |
| return { |
| "detected_patterns": detected_patterns, |
| "pattern_scores": pattern_scores, |
| "overall_pattern_score": min(overall_score, 1.0), |
| "confidence": 0.85 |
| } |
| |
| def get_trace(self) -> Dict[str, Any]: |
| return { |
| "agent": self.name, |
| "version": self.version, |
| "timestamp": "2024-12-31T01:00:00Z", |
| "status": "completed" |
| } |
|
|