Update risk_detector.py

risk_detector.py

@@ -1,30 +1,48 @@
+# risk_detector.py - Enhanced risk detection for legal documents
 from typing import List, Dict, Any
 import re
 
 class RiskDetector:
     def __init__(self):
-        # Define risk keywords/patterns
+        # Define comprehensive risk keywords/patterns
         self.risk_patterns = {
             "penalty_clauses": [
                 r"penalty", r"fine", r"liquidated damages", r"breach of contract",
-                r"default", r"violation", r"non-compliance"
+                r"default", r"violation", r"non-compliance", r"forfeiture",
+                r"damages", r"monetary penalty", r"punitive"
             ],
             "termination_risks": [
                 r"terminate", r"termination", r"cancel", r"cancellation",
-                r"void", r"null and void", r"rescind"
+                r"void", r"null and void", r"rescind", r"dissolution",
+                r"breach", r"expire", r"end", r"cease"
             ],
             "liability_risks": [
                 r"liable", r"liability", r"responsible for", r"accountable",
-                r"damages", r"compensation", r"indemnify"
+                r"damages", r"compensation", r"indemnify", r"hold harmless",
+                r"sue", r"legal action", r"claims", r"losses"
             ],
             "payment_risks": [
                 r"payment default", r"non-payment", r"overdue", r"interest",
-                r"late fee", r"collection", r"debt"
+                r"late fee", r"collection", r"debt", r"delinquent",
+                r"unpaid", r"outstanding", r"arrears"
+            ],
+            "compliance_risks": [
+                r"compliance", r"regulatory", r"legal requirement", r"statute",
+                r"regulation", r"mandatory", r"must comply", r"obligation",
+                r"duty", r"requirement"
+            ],
+            "confidentiality_risks": [
+                r"confidential", r"non-disclosure", r"proprietary", r"trade secret",
+                r"confidentiality", r"disclosure", r"leak", r"unauthorized use"
+            ],
+            "force_majeure": [
+                r"force majeure", r"act of god", r"natural disaster", r"pandemic",
+                r"war", r"emergency", r"unforeseeable", r"beyond control"
             ]
         }
     
     def detect_risks(self, chunks: List[str]) -> List[Dict[str, Any]]:
-        """Detect risky terms across all chunks"""
+        """Detect risky terms across all chunks with enhanced analysis"""
         risk_terms = []
         
         for chunk_idx, chunk in enumerate(chunks):
@@ -34,44 +52,126 @@ class RiskDetector:
                 for pattern in patterns:
                     matches = re.finditer(pattern, chunk_lower, re.IGNORECASE)
                     for match in matches:
-                        # Get context around the match
-                        start = max(0, match.start() - 50)
-                        end = min(len(chunk), match.end() + 50)
+                        # Get context around the match (larger window)
+                        start = max(0, match.start() - 100)
+                        end = min(len(chunk), match.end() + 100)
                         context = chunk[start:end].strip()
                         
+                        # Calculate confidence based on context and pattern
+                        confidence = self._calculate_confidence(pattern, context, chunk)
+                        
                         risk_terms.append({
                             "term": match.group(),
                             "category": risk_category,
                             "context": context,
                             "chunk_index": chunk_idx,
-                            "confidence": self._calculate_confidence(pattern, context)
+                            "confidence": confidence,
+                            "severity": self._assess_severity(risk_category, context),
+                            "position_in_chunk": match.start()
                         })
         
-        # Remove duplicates and sort by confidence
+        # Remove duplicates and sort by confidence and severity
         unique_risks = self._deduplicate_risks(risk_terms)
-        return sorted(unique_risks, key=lambda x: x["confidence"], reverse=True)
+        return sorted(unique_risks, key=lambda x: (x["confidence"], x["severity"]), reverse=True)[:25]  # Top 25 risks
     
-    def _calculate_confidence(self, pattern: str, context: str) -> float:
-        """Simple confidence scoring based on context"""
+    def _calculate_confidence(self, pattern: str, context: str, full_chunk: str) -> float:
+        """Enhanced confidence scoring based on context and surrounding text"""
         confidence = 0.5  # Base confidence
         
-        # Boost confidence for certain patterns
-        high_risk_indicators = ["shall", "must", "required", "obligation"]
+        context_lower = context.lower()
+        
+        # Boost confidence for certain high-risk indicators
+        high_risk_indicators = [
+            "shall", "must", "required", "obligation", "duty", "liable",
+            "penalty", "fine", "terminate", "breach", "void", "damages"
+        ]
+        
+        medium_risk_indicators = [
+            "may", "could", "should", "potential", "possible", "risk"
+        ]
+        
+        # Legal certainty indicators
+        legal_certainty = [
+            "hereby", "whereas", "therefore", "notwithstanding", "pursuant to"
+        ]
+        
         for indicator in high_risk_indicators:
-            if indicator in context.lower():
-                confidence += 0.1
+            if indicator in context_lower:
+                confidence += 0.15
+        
+        for indicator in medium_risk_indicators:
+            if indicator in context_lower:
+                confidence += 0.08
+        
+        for indicator in legal_certainty:
+            if indicator in context_lower:
+                confidence += 0.10
+        
+        # Pattern-specific confidence adjustments
+        if pattern in ["penalty", "fine", "liquidated damages"]:
+            confidence += 0.2  # Financial penalties are high risk
+        elif pattern in ["terminate", "void", "rescind"]:
+            confidence += 0.15  # Contract termination is high risk
+        elif pattern in ["liable", "responsible for", "accountable"]:
+            confidence += 0.12  # Liability is significant risk
+        
+        # Context length penalty (very short contexts are less reliable)
+        if len(context) < 50:
+            confidence -= 0.1
         
         return min(confidence, 1.0)
     
+    def _assess_severity(self, risk_category: str, context: str) -> float:
+        """Assess the severity of the risk based on category and context"""
+        severity_map = {
+            "penalty_clauses": 0.9,
+            "liability_risks": 0.85,
+            "termination_risks": 0.8,
+            "compliance_risks": 0.75,
+            "payment_risks": 0.7,
+            "confidentiality_risks": 0.6,
+            "force_majeure": 0.5
+        }
+        
+        base_severity = severity_map.get(risk_category, 0.5)
+        
+        # Adjust based on context
+        context_lower = context.lower()
+        
+        if any(word in context_lower for word in ["immediate", "immediately", "urgent", "critical"]):
+            base_severity += 0.1
+        
+        if any(word in context_lower for word in ["substantial", "significant", "major", "severe"]):
+            base_severity += 0.08
+        
+        if any(word in context_lower for word in ["minor", "minimal", "limited"]):
+            base_severity -= 0.1
+        
+        return min(base_severity, 1.0)
+    
     def _deduplicate_risks(self, risks: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-        """Remove duplicate risk terms"""
+        """Remove duplicate risk terms with improved logic"""
         seen = set()
         unique_risks = []
         
         for risk in risks:
-            key = (risk["term"].lower(), risk["category"])
+            # Create a more sophisticated key for deduplication
+            key = (
+                risk["term"].lower().strip(),
+                risk["category"],
+                risk["chunk_index"]
+            )
+            
             if key not in seen:
                 seen.add(key)
                 unique_risks.append(risk)
+            else:
+                # If duplicate found, keep the one with higher confidence
+                existing_idx = next(
+                    i for i, r in enumerate(unique_risks)
+                    if (r["term"].lower().strip(), r["category"], r["chunk_index"]) == key
+                )
+                if risk["confidence"] > unique_risks[existing_idx]["confidence"]:
+                    unique_risks[existing_idx] = risk
         
         return unique_risks