| | """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" |
| | } |
| |
|