How AI Powers SOC Operations in 2026

python3 -m venv .venv-ai-soc
source .venv-ai-soc/bin/activate
pip install --upgrade pip
pip install pandas

cat > alerts.csv <<'CSV'
id,source,severity,asset_criticality,description
a1,edr,high,prod,"outbound to 198.51.100.10"
a2,waf,medium,prod,"multiple 404 bursts"
a3,siem,low,dev,"failed logins from 10.0.0.9"
CSV

cat > ai_soc_assistant.py <<'PY'
#!/usr/bin/env python3
"""
AI-Powered SOC Assistant
Uses LLM to triage alerts, generate responses, and assist incident response
"""

import pandas as pd
import json
import os
from datetime import datetime
from typing import List, Dict, Optional
import hashlib

# For OpenAI API (or use local LLM)
try:
    import openai
    OPENAI_AVAILABLE = True
except ImportError:
    OPENAI_AVAILABLE = False
    print("OpenAI not available. Using rule-based fallback.")

class AlertNormalizer:
    """Normalize alerts from different sources"""
    
    @staticmethod
    def normalize_alert(alert: Dict) -> Dict:
        """Normalize alert to standard format"""
        normalized = {
            'id': alert.get('id', ''),
            'source': alert.get('source', 'unknown'),
            'severity': alert.get('severity', 'low').lower(),
            'asset_criticality': alert.get('asset_criticality', 'unknown').lower(),
            'description': alert.get('description', ''),
            'timestamp': alert.get('timestamp', datetime.now().isoformat()),
            'src_ip': alert.get('src_ip', ''),
            'dst_ip': alert.get('dst_ip', ''),
            'user': alert.get('user', ''),
            'action': alert.get('action', ''),
        }
        return normalized

class RiskScorer:
    """Calculate risk scores for alerts"""
    
    SEVERITY_WEIGHTS = {
        'critical': 20,
        'high': 10,
        'medium': 5,
        'low': 1
    }
    
    CRITICALITY_WEIGHTS = {
        'prod': 10,
        'staging': 5,
        'dev': 2,
        'test': 1
    }
    
    def calculate_score(self, alert: Dict) -> float:
        """Calculate risk score"""
        base_score = self.SEVERITY_WEIGHTS.get(alert['severity'], 1)
        criticality_bonus = self.CRITICALITY_WEIGHTS.get(alert['asset_criticality'], 1)
        
        # Additional factors
        external_ip_bonus = 3 if alert.get('src_ip', '').startswith('198.51.100') else 0
        failed_action_bonus = 2 if 'fail' in alert.get('status', '').lower() else 0
        
        total_score = base_score + criticality_bonus + external_ip_bonus + failed_action_bonus
        return min(total_score, 100)  # Cap at 100

class AISOCAssistant:
    """AI-powered SOC assistant using LLM"""
    
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.getenv('OPENAI_API_KEY')
        self.use_llm = OPENAI_AVAILABLE and self.api_key
        if self.use_llm:
            openai.api_key = self.api_key
        self.alert_history = []
    
    def generate_response(self, alert: Dict, context: List[Dict] = None) -> Dict:
        """Generate AI response for alert"""
        if not self.use_llm:
            return self._rule_based_response(alert)
        
        # Build prompt for LLM
        prompt = self._build_prompt(alert, context)
        
        try:
            response = openai.ChatCompletion.create(
                model="gpt-4",
                messages=[
                    {"role": "system", "content": self._get_system_prompt()},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3,
                max_tokens=500
            )
            
            ai_response = response.choices[0].message.content
            return self._parse_ai_response(ai_response, alert)
        
        except Exception as e:
            print(f"LLM error: {e}, falling back to rule-based")
            return self._rule_based_response(alert)
    
    def _get_system_prompt(self) -> str:
        """System prompt for security context"""
        return """You are an AI security operations assistant. Your role is to:
1. Analyze security alerts and assess risk
2. Suggest appropriate response actions
3. Correlate alerts with known attack patterns
4. Provide clear, actionable recommendations

Always prioritize:
- Production systems and critical assets
- External IP addresses and unknown sources
- Failed authentication attempts
- Unusual network activity

Be concise and specific in your recommendations."""
    
    def _build_prompt(self, alert: Dict, context: List[Dict] = None) -> str:
        """Build prompt for LLM"""
        prompt = f"""Analyze this security alert and provide recommendations:

Alert Details:
- ID: {alert['id']}
- Source: {alert['source']}
- Severity: {alert['severity']}
- Asset: {alert['asset_criticality']}
- Description: {alert['description']}
- Source IP: {alert.get('src_ip', 'N/A')}
- Timestamp: {alert['timestamp']}
"""
        
        if context:
            prompt += "\nRelated Alerts:\n"
            for ctx_alert in context[:3]:
                prompt += f"- {ctx_alert['description']}\n"
        
        prompt += "\nProvide:\n1. Risk assessment (1-10)\n2. Recommended actions (3-5 steps)\n3. Priority level\n4. Suggested investigation steps"
        
        return prompt
    
    def _parse_ai_response(self, ai_text: str, alert: Dict) -> Dict:
        """Parse AI response into structured format"""
        return {
            'risk_assessment': self._extract_risk(ai_text),
            'recommended_actions': self._extract_actions(ai_text),
            'priority': self._extract_priority(ai_text),
            'investigation_steps': self._extract_steps(ai_text),
            'ai_summary': ai_text[:200] + "..." if len(ai_text) > 200 else ai_text
        }
    
    def _extract_risk(self, text: str) -> int:
        """Extract risk score from AI response"""
        import re
        match = re.search(r'risk[:\s]+(\d+)', text, re.IGNORECASE)
        return int(match.group(1)) if match else 5
    
    def _extract_actions(self, text: str) -> List[str]:
        """Extract recommended actions"""
        actions = []
        lines = text.split('\n')
        for line in lines:
            if any(keyword in line.lower() for keyword in ['contain', 'block', 'isolate', 'investigate', 'reset']):
                actions.append(line.strip('- ').strip())
        return actions[:5] if actions else ["Review alert", "Correlate with other events", "Decide on response"]
    
    def _extract_priority(self, text: str) -> str:
        """Extract priority level"""
        text_lower = text.lower()
        if 'critical' in text_lower or 'immediate' in text_lower:
            return 'CRITICAL'
        elif 'high' in text_lower:
            return 'HIGH'
        elif 'medium' in text_lower:
            return 'MEDIUM'
        else:
            return 'LOW'
    
    def _extract_steps(self, text: str) -> List[str]:
        """Extract investigation steps"""
        steps = []
        lines = text.split('\n')
        for line in lines:
            if any(keyword in line.lower() for keyword in ['check', 'review', 'analyze', 'correlate', 'verify']):
                steps.append(line.strip('- ').strip())
        return steps[:5] if steps else ["Review alert details", "Check related events", "Verify asset status"]
    
    def _rule_based_response(self, alert: Dict) -> Dict:
        """Rule-based fallback when LLM unavailable"""
        actions = []
        steps = []
        risk = 5
        
        desc_lower = alert['description'].lower()
        
        if 'outbound' in desc_lower or 'exfiltration' in desc_lower:
            actions = [
                "Contain host (analyst approval required)",
                "Block destination IP/domain",
                "Collect network logs (PCAP)",
                "Open incident ticket"
            ]
            steps = [
                "Verify outbound connection legitimacy",
                "Check destination IP reputation",
                "Review data transfer volume",
                "Correlate with user activity"
            ]
            risk = 8
        
        elif 'login' in desc_lower or 'authentication' in desc_lower:
            actions = [
                "Reset credentials (after validation)",
                "Check MFA status",
                "Review access logs",
                "Notify user"
            ]
            steps = [
                "Verify failed login attempts",
                "Check for credential stuffing patterns",
                "Review account activity history",
                "Validate user identity"
            ]
            risk = 6
        
        elif 'burst' in desc_lower or 'flood' in desc_lower:
            actions = [
                "Rate limit source IP",
                "Check for DDoS attack",
                "Review application logs",
                "Scale resources if needed"
            ]
            steps = [
                "Analyze traffic patterns",
                "Check for attack signatures",
                "Review resource utilization",
                "Correlate with network events"
            ]
            risk = 7
        
        else:
            actions = [
                "Review alert context",
                "Correlate with asset owner",
                "Decide on containment",
                "Document findings"
            ]
            steps = [
                "Review alert details",
                "Check related events",
                "Verify asset criticality",
                "Assess business impact"
            ]
            risk = 4
        
        return {
            'risk_assessment': risk,
            'recommended_actions': actions,
            'priority': alert['severity'].upper(),
            'investigation_steps': steps,
            'ai_summary': f"Rule-based analysis: {alert['description']}"
        }
    
    def correlate_alerts(self, alerts: List[Dict]) -> List[Dict]:
        """Correlate multiple alerts to identify attack patterns"""
        correlated = []
        
        # Group by source IP
        ip_groups = {}
        for alert in alerts:
            src_ip = alert.get('src_ip', 'unknown')
            if src_ip not in ip_groups:
                ip_groups[src_ip] = []
            ip_groups[src_ip].append(alert)
        
        # Identify patterns
        for src_ip, ip_alerts in ip_groups.items():
            if len(ip_alerts) >= 3:
                correlated.append({
                    'pattern': 'multiple_alerts_from_same_ip',
                    'src_ip': src_ip,
                    'alert_count': len(ip_alerts),
                    'alerts': ip_alerts,
                    'recommendation': 'Potential coordinated attack or compromised host'
                })
        
        return correlated

def main():
    # Load alerts
df = pd.read_csv("alerts.csv")

    # Initialize components
    normalizer = AlertNormalizer()
    scorer = RiskScorer()
    assistant = AISOCAssistant()
    
    # Process alerts
    results = []
    all_alerts = []
    
    for _, row in df.iterrows():
        alert = normalizer.normalize_alert(row.to_dict())
        score = scorer.calculate_score(alert)
        alert['risk_score'] = score
        
        # Get AI response
        response = assistant.generate_response(alert)
        alert.update(response)
        
        all_alerts.append(alert)
        results.append(alert)
    
    # Correlate alerts
    correlations = assistant.correlate_alerts(all_alerts)
    
    # Create results DataFrame
    results_df = pd.DataFrame(results)
    results_df = results_df.sort_values('risk_score', ascending=False)
    
    # Save results
    results_df.to_csv("alerts_triaged.csv", index=False)
    
    # Save correlations
    with open("correlations.json", 'w') as f:
        json.dump(correlations, f, indent=2, default=str)
    
    # Print summary
    print("\n" + "="*60)
    print("AI SOC Assistant - Triage Results")
    print("="*60)
    print(f"Total alerts processed: {len(results)}")
    print(f"High priority alerts: {len(results_df[results_df['risk_score'] >= 15])}")
    print(f"Correlations found: {len(correlations)}")
    
    print("\nTop Priority Alerts:")
    for _, alert in results_df.head(5).iterrows():
        print(f"\n  [{alert['priority']}] {alert['id']}: {alert['description']}")
        print(f"     Risk Score: {alert['risk_score']}")
        print(f"     Actions: {', '.join(alert['recommended_actions'][:2])}")
    
    if correlations:
        print("\n\nAlert Correlations:")
        for corr in correlations:
            print(f"  Pattern: {corr['pattern']}")
            print(f"  Source IP: {corr['src_ip']}")
            print(f"  Alert Count: {corr['alert_count']}")
            print(f"  Recommendation: {corr['recommendation']}")

if __name__ == "__main__":
    main()
PY

# Install OpenAI if using LLM (optional)
# pip install openai

python ai_soc_assistant.py

cat > gpt_soc_assistant.py <<'PY'
#!/usr/bin/env python3
"""
GPT-Based SOC Assistant
Automatically summarizes alerts and suggests response actions
"""

import os
import json
import pandas as pd
from datetime import datetime
from typing import List, Dict, Optional
import openai
from pathlib import Path

class GPTBasedSOCAssistant:
    """GPT-powered SOC assistant for alert summarization"""
    
    def __init__(self, api_key: Optional[str] = None, model: str = "gpt-4"):
        self.api_key = api_key or os.getenv('OPENAI_API_KEY')
        self.model = model
        if self.api_key:
            openai.api_key = self.api_key
        self.alert_history = []
        self.summaries = []
    
    def summarize_alerts(self, alerts: List[Dict]) -> Dict:
        """Summarize multiple alerts using GPT"""
        if not self.api_key:
            return self._rule_based_summary(alerts)
        
        # Prepare alert context
        alert_context = self._prepare_alert_context(alerts)
        
        # Build prompt for GPT
        prompt = f"""You are a Security Operations Center (SOC) analyst assistant. 
Analyze the following security alerts and provide a comprehensive summary.

Alerts:
{alert_context}

Provide:
1. Executive Summary (2-3 sentences)
2. Risk Assessment (Overall risk level: LOW/MEDIUM/HIGH/CRITICAL)
3. Top 3 Priority Alerts (with reasons)
4. Recommended Immediate Actions (3-5 steps)
5. Investigation Steps (detailed steps for each priority alert)
6. Potential Attack Pattern (if multiple alerts correlate)

Format your response as JSON with these keys:
- executive_summary
- risk_level
- priority_alerts (list of alert IDs with reasons)
- immediate_actions (list)
- investigation_steps (dict with alert_id as key)
- attack_pattern (string or null)
"""
        
        try:
            response = openai.ChatCompletion.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "You are an expert SOC analyst assistant. Provide clear, actionable security recommendations."},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3,
                max_tokens=1500
            )
            
            ai_response = response.choices[0].message.content
            
            # Parse JSON response
            try:
                summary = json.loads(ai_response)
            except json.JSONDecodeError:
                # If not valid JSON, extract structured info
                summary = self._parse_text_response(ai_response, alerts)
            
            summary['timestamp'] = datetime.now().isoformat()
            summary['alert_count'] = len(alerts)
            summary['model_used'] = self.model
            
            self.summaries.append(summary)
            return summary
        
        except Exception as e:
            print(f"GPT API error: {e}, falling back to rule-based")
            return self._rule_based_summary(alerts)
    
    def _prepare_alert_context(self, alerts: List[Dict]) -> str:
        """Prepare alert context for GPT"""
        context_lines = []
        for i, alert in enumerate(alerts[:20], 1):  # Limit to 20 alerts
            context_lines.append(f"""
Alert {i}:
- ID: {alert.get('id', 'N/A')}
- Source: {alert.get('source', 'N/A')}
- Severity: {alert.get('severity', 'N/A')}
- Asset: {alert.get('asset_criticality', 'N/A')}
- Description: {alert.get('description', 'N/A')}
- Source IP: {alert.get('src_ip', 'N/A')}
- Timestamp: {alert.get('timestamp', 'N/A')}
""")
        return '\n'.join(context_lines)
    
    def _parse_text_response(self, text: str, alerts: List[Dict]) -> Dict:
        """Parse text response if JSON parsing fails"""
        summary = {
            'executive_summary': text[:200] + "...",
            'risk_level': 'MEDIUM',
            'priority_alerts': [],
            'immediate_actions': [],
            'investigation_steps': {},
            'attack_pattern': None
        }
        
        # Extract priority alerts (first 3 by severity)
        sorted_alerts = sorted(alerts, key=lambda x: self._severity_to_num(x.get('severity', 'low')), reverse=True)
        summary['priority_alerts'] = [
            {'id': a.get('id'), 'reason': f"High severity: {a.get('severity')}"}
            for a in sorted_alerts[:3]
        ]
        
        return summary
    
    def _severity_to_num(self, severity: str) -> int:
        """Convert severity to number for sorting"""
        severity_map = {'critical': 4, 'high': 3, 'medium': 2, 'low': 1}
        return severity_map.get(severity.lower(), 0)
    
    def _rule_based_summary(self, alerts: List[Dict]) -> Dict:
        """Rule-based fallback summary"""
        total_alerts = len(alerts)
        critical_count = sum(1 for a in alerts if a.get('severity', '').lower() == 'critical')
        high_count = sum(1 for a in alerts if a.get('severity', '').lower() == 'high')
        
        # Determine risk level
        if critical_count > 0:
            risk_level = 'CRITICAL'
        elif high_count > 5:
            risk_level = 'HIGH'
        elif high_count > 0:
            risk_level = 'MEDIUM'
        else:
            risk_level = 'LOW'
        
        # Get priority alerts
        sorted_alerts = sorted(alerts, key=lambda x: self._severity_to_num(x.get('severity', 'low')), reverse=True)
        priority_alerts = [
            {'id': a.get('id'), 'reason': f"{a.get('severity', 'unknown')} severity alert"}
            for a in sorted_alerts[:3]
        ]
        
        # Generate actions based on alert types
        immediate_actions = []
        if critical_count > 0:
            immediate_actions.append("Immediately investigate critical alerts")
        if high_count > 0:
            immediate_actions.append("Review high-severity alerts within 1 hour")
        immediate_actions.append("Correlate alerts with recent security events")
        immediate_actions.append("Check for related incidents in ticketing system")
        
        return {
            'executive_summary': f"Received {total_alerts} security alerts. {critical_count} critical, {high_count} high severity alerts require immediate attention.",
            'risk_level': risk_level,
            'priority_alerts': priority_alerts,
            'immediate_actions': immediate_actions,
            'investigation_steps': {
                a.get('id'): [f"Review {a.get('source')} alert", "Check asset status", "Verify with asset owner"]
                for a in sorted_alerts[:3]
            },
            'attack_pattern': None,
            'timestamp': datetime.now().isoformat(),
            'alert_count': total_alerts,
            'model_used': 'rule-based'
        }
    
    def generate_incident_report(self, alerts: List[Dict]) -> str:
        """Generate formatted incident report"""
        summary = self.summarize_alerts(alerts)
        
        report = f"""
╔══════════════════════════════════════════════════════════════╗
║           SOC ALERT SUMMARY REPORT                            ║
║           Generated: {summary['timestamp']}                  ║
╚══════════════════════════════════════════════════════════════╝

EXECUTIVE SUMMARY
─────────────────
{summary['executive_summary']}

RISK ASSESSMENT
───────────────
Overall Risk Level: {summary['risk_level']}
Total Alerts: {summary['alert_count']}

PRIORITY ALERTS
───────────────
"""
        for i, alert in enumerate(summary['priority_alerts'], 1):
            report += f"{i}. Alert {alert['id']}: {alert['reason']}\n"
        
        report += f"""
IMMEDIATE ACTIONS
─────────────────
"""
        for i, action in enumerate(summary['immediate_actions'], 1):
            report += f"{i}. {action}\n"
        
        if summary.get('attack_pattern'):
            report += f"""
POTENTIAL ATTACK PATTERN
────────────────────────
{summary['attack_pattern']}
"""
        
        return report
    
    def save_summary(self, filepath: str):
        """Save summary to file"""
        if self.summaries:
            with open(filepath, 'w') as f:
                json.dump(self.summaries[-1], f, indent=2, default=str)
            print(f"Summary saved to {filepath}")

def main():
    # Load alerts
    df = pd.read_csv("alerts.csv")
    
    # Convert to list of dicts
    alerts = df.to_dict('records')
    
    # Initialize GPT assistant
    assistant = GPTBasedSOCAssistant()
    
    print("="*60)
    print("GPT-Based SOC Assistant")
    print("="*60)
    print(f"Processing {len(alerts)} alerts...\n")
    
    # Generate summary
    summary = assistant.summarize_alerts(alerts)
    
    # Generate report
    report = assistant.generate_incident_report(alerts)
    print(report)
    
    # Save summary
    assistant.save_summary("soc_alert_summary.json")
    
    print("\n" + "="*60)
    print("Summary Complete!")
    print("="*60)

if __name__ == "__main__":
    main()
PY

# Install OpenAI
pip install openai

# Run GPT SOC Assistant
python gpt_soc_assistant.py

cat > siem_integration.py <<'PY'
#!/usr/bin/env python3
"""
SIEM Integration for GPT SOC Assistant
Integrates with Splunk, Elasticsearch, or custom SIEM
"""

import requests
import json
from datetime import datetime, timedelta
from typing import List, Dict

class SIEMIntegration:
    """Integrate with SIEM systems"""
    
    def __init__(self, siem_type: str = 'elasticsearch', config: Dict = None):
        self.siem_type = siem_type
        self.config = config or {}
    
    def fetch_alerts_from_elasticsearch(self, time_range_minutes: int = 60) -> List[Dict]:
        """Fetch alerts from Elasticsearch"""
        es_url = self.config.get('url', 'http://localhost:9200')
        index = self.config.get('index', 'security-alerts')
        
        # Query for recent alerts
        query = {
            "query": {
                "range": {
                    "@timestamp": {
                        "gte": f"now-{time_range_minutes}m"
                    }
                }
            },
            "size": 100,
            "sort": [{"@timestamp": {"order": "desc"}}]
        }
        
        try:
            response = requests.post(
                f"{es_url}/{index}/_search",
                json=query,
                auth=(self.config.get('username'), self.config.get('password')) if self.config.get('username') else None
            )
            response.raise_for_status()
            
            hits = response.json().get('hits', {}).get('hits', [])
            alerts = []
            
            for hit in hits:
                source = hit.get('_source', {})
                alerts.append({
                    'id': hit.get('_id'),
                    'source': 'elasticsearch',
                    'severity': source.get('severity', 'medium'),
                    'description': source.get('message', ''),
                    'timestamp': source.get('@timestamp'),
                    'src_ip': source.get('source', {}).get('ip'),
                    'asset_criticality': source.get('asset', {}).get('criticality', 'unknown')
                })
            
            return alerts
        
        except Exception as e:
            print(f"Error fetching from Elasticsearch: {e}")
            return []
    
    def fetch_alerts_from_splunk(self, time_range_minutes: int = 60) -> List[Dict]:
        """Fetch alerts from Splunk"""
        splunk_url = self.config.get('url', 'https://localhost:8089')
        search_query = f'index=security earliest=-{time_range_minutes}m | head 100'
        
        try:
            # Splunk REST API call (simplified)
            response = requests.post(
                f"{splunk_url}/services/search/jobs/oneshot",
                data={'search': search_query, 'output_mode': 'json'},
                auth=(self.config.get('username'), self.config.get('password')),
                verify=False  # In production, use proper certificates
            )
            response.raise_for_status()
            
            results = response.json().get('results', [])
            alerts = []
            
            for result in results:
                alerts.append({
                    'id': result.get('_cd'),
                    'source': 'splunk',
                    'severity': result.get('severity', 'medium'),
                    'description': result.get('_raw', ''),
                    'timestamp': result.get('_time'),
                    'src_ip': result.get('src_ip'),
                    'asset_criticality': result.get('asset_criticality', 'unknown')
                })
            
            return alerts
        
        except Exception as e:
            print(f"Error fetching from Splunk: {e}")
            return []
    
    def create_ticket(self, alert_summary: Dict, ticketing_system: str = 'jira') -> str:
        """Create ticket in ticketing system"""
        if ticketing_system == 'jira':
            jira_url = self.config.get('jira_url', 'https://your-domain.atlassian.net')
            project_key = self.config.get('project_key', 'SOC')
            
            ticket_data = {
                "fields": {
                    "project": {"key": project_key},
                    "summary": f"SOC Alert Summary - {alert_summary['risk_level']} Risk",
                    "description": alert_summary['executive_summary'],
                    "issuetype": {"name": "Incident"},
                    "priority": {"name": alert_summary['risk_level']}
                }
            }
            
            try:
                response = requests.post(
                    f"{jira_url}/rest/api/2/issue",
                    json=ticket_data,
                    auth=(self.config.get('jira_username'), self.config.get('jira_token')),
                    headers={"Content-Type": "application/json"}
                )
                response.raise_for_status()
                ticket_key = response.json().get('key')
                print(f"Ticket created: {ticket_key}")
                return ticket_key
            except Exception as e:
                print(f"Error creating Jira ticket: {e}")
                return None
        
        return None

# Usage example
if __name__ == "__main__":
    from gpt_soc_assistant import GPTBasedSOCAssistant
    
    # Initialize SIEM integration
    siem = SIEMIntegration('elasticsearch', {
        'url': 'http://localhost:9200',
        'index': 'security-alerts'
    })
    
    # Fetch alerts
    alerts = siem.fetch_alerts_from_elasticsearch(time_range_minutes=60)
    
    # Process with GPT assistant
    assistant = GPTBasedSOCAssistant()
    summary = assistant.summarize_alerts(alerts)
    
    # Create ticket if high risk
    if summary['risk_level'] in ['HIGH', 'CRITICAL']:
        siem.create_ticket(summary, 'jira')
PY

python siem_integration.py

import hashlib
import json
from datetime import datetime
from typing import Dict, List

class SOCValidation:
    """Validation controls for AI SOC operations"""
    
    def __init__(self):
        self.analyst_decisions = {}
        self.ai_decisions = {}
        self.metrics = {
            "true_positives": 0,
            "false_positives": 0,
            "false_negatives": 0,
            "true_negatives": 0
        }
    
    def hash_alert(self, alert: Dict) -> str:
        """Hash alert for integrity checking"""
        alert_str = json.dumps(alert, sort_keys=True)
        return hashlib.sha256(alert_str.encode()).hexdigest()
    
    def record_decision(self, alert_id: str, source: str, decision: str, score: float):
        """Record decision from AI or analyst"""
        if source == "ai":
            self.ai_decisions[alert_id] = {"decision": decision, "score": score}
        else:
            self.analyst_decisions[alert_id] = {"decision": decision}
    
    def calculate_metrics(self) -> Dict:
        """Calculate precision/recall metrics"""
        for alert_id in self.ai_decisions:
            ai_decision = self.ai_decisions[alert_id]["decision"]
            analyst_decision = self.analyst_decisions.get(alert_id, {}).get("decision")
            
            if analyst_decision:
                if ai_decision == "investigate" and analyst_decision == "investigate":
                    self.metrics["true_positives"] += 1
                elif ai_decision == "investigate" and analyst_decision == "ignore":
                    self.metrics["false_positives"] += 1
                elif ai_decision == "ignore" and analyst_decision == "investigate":
                    self.metrics["false_negatives"] += 1
                else:
                    self.metrics["true_negatives"] += 1
        
        tp = self.metrics["true_positives"]
        fp = self.metrics["false_positives"]
        fn = self.metrics["false_negatives"]
        
        precision = tp / (tp + fp) if (tp + fp) > 0 else 0
        recall = tp / (tp + fn) if (tp + fn) > 0 else 0
        f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
        
        return {
            "precision": precision,
            "recall": recall,
            "f1": f1,
            "metrics": self.metrics
        }

deactivate || true
rm -rf .venv-ai-soc alerts.csv alerts_triaged.csv triage.py