{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/Hawksight-AI/semantica/blob/main/cookbook/use_cases/cybersecurity/01_Anomaly_Detection_Real_Time.ipynb)\n",
    "\n",
    "# Real-Time Anomaly Detection Pipeline\n",
    "\n",
    "## Overview\n",
    "\n",
    "This notebook demonstrates a complete real-time anomaly detection pipeline for cybersecurity: stream security logs from multiple sources, parse in real-time, build temporal knowledge graph, detect anomalies using pattern detection and inference, generate alerts, and monitor continuously.\n",
    "\n",
    "\n",
    "**Documentation**: [API Reference](https://semantica.readthedocs.io/use-cases/)\n",
    "\n",
    "## Installation\n",
    "\n",
    "Install Semantica from PyPI:\n",
    "\n",
    "```bash\n",
    "pip install semantica\n",
    "# Or with all optional dependencies:\n",
    "pip install semantica[all]\n",
    "```\n",
    "\n",
    "### Modules Used (20+)\n",
    "\n",
    "- **Ingestion**: FileIngestor, WebIngestor, FeedIngestor, StreamIngestor, DBIngestor, EmailIngestor, RepoIngestor, MCPIngestor\n",
    "- **Parsing**: JSONParser, StructuredDataParser, DocumentParser\n",
    "- **Extraction**: NERExtractor, RelationExtractor, EventDetector, TripleExtractor\n",
    "- **KG**: GraphBuilder, TemporalPatternDetector, TemporalGraphQuery, GraphAnalyzer\n",
    "- **Analytics**: CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n",
    "- **Reasoning**: InferenceEngine, RuleManager, ExplanationGenerator\n",
    "- **Quality**: KGQualityAssessor, AutomatedFixer\n",
    "- **Export**: JSONExporter, CSVExporter, ReportGenerator\n",
    "- **Visualization**: KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
    "\n",
    "### Pipeline\n",
    "\n",
    "**Stream Security Logs → Real-Time Parsing → Extract Entities → Build Temporal KG → Pattern Detection → Anomaly Detection → Generate Alerts → Monitor → Visualize**\n",
    "\n",
    "---\n",
    "\n",
    "## Step 1: Stream Security Logs from Multiple Sources\n",
    "\n",
    "Stream security logs from files, databases, and real-time sources.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from semantica.ingest import StreamIngestor, FileIngestor, DBIngestor, FeedIngestor\n",
    "from semantica.parse import JSONParser, StructuredDataParser, DocumentParser\n",
    "from semantica.semantic_extract import NERExtractor, RelationExtractor, EventDetector, TripleExtractor\n",
    "from semantica.kg import GraphBuilder, TemporalPatternDetector, TemporalGraphQuery, GraphAnalyzer\n",
    "from semantica.kg import CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n",
    "from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
    
    "from semantica.export import JSONExporter, CSVExporter, ReportGenerator\n",
    "from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
    "import tempfile\n",
    "import os\n",
    "import json\n",
    "import time\n",
    "from datetime import datetime, timedelta\n",
    "from collections import deque\n",
    "\n",
    "stream_ingestor = StreamIngestor()\n",
    "file_ingestor = FileIngestor()\n",
    "db_ingestor = DBIngestor()\n",
    "feed_ingestor = FeedIngestor()\n",
    "\n",
    "# Real streaming sources configuration\n",
    "stream_sources = [\n",
    "    {\n",
    "        \"type\": \"kafka\",\n",
    "        \"topic\": \"security_logs\",\n",
    "        \"bootstrap_servers\": [\"localhost:9092\"],\n",
    "        \"consumer_config\": {\"group_id\": \"semantica_security_monitor\"}\n",
    "    },\n",
    "    {\n",
    "        \"type\": \"rabbitmq\",\n",
    "        \"queue\": \"security_events\",\n",
    "        \"connection_url\": \"amqp://user:password@localhost:5672/\"\n",
    "    }\n",
    "]\n",
    "\n",
    "# Real database connection for security logs\n",
    "db_connection_string = \"postgresql://user:password@localhost:5432/security_logs_db\"\n",
    "db_query = \"SELECT * FROM security_events WHERE timestamp > NOW() - INTERVAL '1 hour' ORDER BY timestamp DESC LIMIT 1000\"\n",
    "\n",
    "# Real security feed URLs for threat intelligence\n",
    "security_feeds = [\n",
    "    \"https://www.cisa.gov/news.xml\",\n",
    "    \"https://www.us-cert.gov/ncas/alerts.xml\"\n",
    "]\n",
    "\n",
    "json_parser = JSONParser()\n",
    "structured_parser = StructuredDataParser()\n",
    "document_parser = DocumentParser()\n",
    "\n",
    "temp_dir = tempfile.mkdtemp()\n",
    "\n",
    "# Real-world streaming security log format (simulating real-time stream)\n",
    "security_log_stream_file = os.path.join(temp_dir, \"security_log_stream.json\")\n",
    "stream_logs = [\n",
    "    {\n",
    "        \"timestamp\": (datetime.now() - timedelta(minutes=5)).isoformat(),\n",
    "        \"source_ip\": \"192.168.1.50\",\n",
    "        \"destination_ip\": \"10.0.0.100\",\n",
    "        \"event_type\": \"normal_traffic\",\n",
    "        \"bytes_sent\": 1024,\n",
    "        \"bytes_received\": 2048,\n",
    "        \"protocol\": \"TCP\",\n",
    "        \"port\": 80\n",
    "    },\n",
    "    {\n",
    "        \"timestamp\": (datetime.now() - timedelta(minutes=4)).isoformat(),\n",
    "        \"source_ip\": \"203.0.113.100\",\n",
    "        \"destination_ip\": \"10.0.0.100\",\n",
    "        \"event_type\": \"suspicious_connection\",\n",
    "        \"bytes_sent\": 5000000,\n",
    "        \"bytes_received\": 1000,\n",
    "        \"protocol\": \"TCP\",\n",
    "        \"port\": 443\n",
    "    },\n",
    "    {\n",
    "        \"timestamp\": (datetime.now() - timedelta(minutes=3)).isoformat(),\n",
    "        \"source_ip\": \"192.168.1.50\",\n",
    "        \"destination_ip\": \"10.0.0.100\",\n",
    "        \"event_type\": \"normal_traffic\",\n",
    "        \"bytes_sent\": 512,\n",
    "        \"bytes_received\": 1024,\n",
    "        \"protocol\": \"UDP\",\n",
    "        \"port\": 53\n",
    "    },\n",
    "    {\n",
    "        \"timestamp\": (datetime.now() - timedelta(minutes=2)).isoformat(),\n",
    "        \"source_ip\": \"198.51.100.50\",\n",
    "        \"destination_ip\": \"10.0.0.100\",\n",
    "        \"event_type\": \"port_scan\",\n",
    "        \"bytes_sent\": 100,\n",
    "        \"bytes_received\": 0,\n",
    "        \"protocol\": \"TCP\",\n",
    "        \"port\": 22\n",
    "    },\n",
    "    {\n",
    "        \"timestamp\": (datetime.now() - timedelta(minutes=1)).isoformat(),\n",
    "        \"source_ip\": \"203.0.113.100\",\n",
    "        \"destination_ip\": \"10.0.0.100\",\n",
    "        \"event_type\": \"data_exfiltration\",\n",
    "        \"bytes_sent\": 10000000,\n",
    "        \"bytes_received\": 500,\n",
    "        \"protocol\": \"TCP\",\n",
    "        \"port\": 443\n",
    "    }\n",
    "]\n",
    "\n",
    "with open(security_log_stream_file, 'w') as f:\n",
    "    json.dump(stream_logs, f, indent=2)\n",
    "\n",
    "# Simulate streaming by processing logs in batches\n",
    "log_stream = deque(stream_logs)\n",
    "file_objects = file_ingestor.ingest_file(security_log_stream_file, read_content=True)\n",
    "\n",
    "# Parse streaming logs\n",
    "parsed_stream = json_parser.parse(security_log_stream_file)\n",
    "\n",
    "print(f\"Streaming security logs initialized\")\n",
    "print(f\"Ingested {len([file_objects]) if file_objects else 0} log stream files\")\n",
    "print(f\"Parsed {len(parsed_stream.data) if parsed_stream and parsed_stream.data else 0} log entries\")\n",
    "print(f\"Stream ready for real-time processing\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Real-Time Parsing and Entity Extraction\n",
    "\n",
    "Parse streaming logs in real-time and extract security entities.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ner_extractor = NERExtractor()\n",
    "relation_extractor = RelationExtractor()\n",
    "event_detector = EventDetector()\n",
    "triple_extractor = TripleExtractor()\n",
    "\n",
    "# Real-time processing loop (simulated)\n",
    "security_entities = []\n",
    "stream_relationships = []\n",
    "detected_events = []\n",
    "\n",
    "# Process logs in real-time batches\n",
    "for log_entry in parsed_stream.data if parsed_stream and parsed_stream.data else []:\n",
    "    if isinstance(log_entry, dict):\n",
    "        log_text = f\"{log_entry.get('event_type', '')} from {log_entry.get('source_ip', '')} to {log_entry.get('destination_ip', '')} on port {log_entry.get('port', '')}\"\n",
    "        \n",
    "        entities = ner_extractor.extract(log_text)\n",
    "        relationships = relation_extractor.extract(log_text, entities)\n",
    "        events = event_detector.detect_events(log_text)\n",
    "        \n",
    "        security_entities.append({\n",
    "            \"id\": log_entry.get(\"source_ip\", \"\"),\n",
    "            \"type\": \"IP_Address\",\n",
    "            \"name\": log_entry.get(\"source_ip\", \"\"),\n",
    "            \"properties\": {\n",
    "                \"timestamp\": log_entry.get(\"timestamp\", \"\"),\n",
    "                \"source\": \"stream\"\n",
    "            }\n",
    "        })\n",
    "        security_entities.append({\n",
    "            \"id\": log_entry.get(\"destination_ip\", \"\"),\n",
    "            \"type\": \"IP_Address\",\n",
    "            \"name\": log_entry.get(\"destination_ip\", \"\"),\n",
    "            \"properties\": {\n",
    "                \"timestamp\": log_entry.get(\"timestamp\", \"\"),\n",
    "                \"source\": \"stream\"\n",
    "            }\n",
    "        })\n",
    "        security_entities.append({\n",
    "            \"id\": log_entry.get(\"event_type\", \"\"),\n",
    "            \"type\": \"Security_Event\",\n",
    "            \"name\": log_entry.get(\"event_type\", \"\"),\n",
    "            \"properties\": {\n",
    "                \"timestamp\": log_entry.get(\"timestamp\", \"\"),\n",
    "                \"bytes_sent\": log_entry.get(\"bytes_sent\", 0),\n",
    "                \"bytes_received\": log_entry.get(\"bytes_received\", 0),\n",
    "                \"protocol\": log_entry.get(\"protocol\", \"\"),\n",
    "                \"port\": log_entry.get(\"port\", 0)\n",
    "            }\n",
    "        })\n",
    "        \n",
    "        stream_relationships.append({\n",
    "            \"source\": log_entry.get(\"source_ip\", \"\"),\n",
    "            \"target\": log_entry.get(\"event_type\", \"\"),\n",
    "            \"type\": \"triggered\",\n",
    "            \"properties\": {\"timestamp\": log_entry.get(\"timestamp\", \"\")}\n",
    "        })\n",
    "        stream_relationships.append({\n",
    "            \"source\": log_entry.get(\"event_type\", \"\"),\n",
    "            \"target\": log_entry.get(\"destination_ip\", \"\"),\n",
    "            \"type\": \"targeted\",\n",
    "            \"properties\": {\"timestamp\": log_entry.get(\"timestamp\", \"\")}\n",
    "        })\n",
    "        \n",
    "        detected_events.extend(events)\n",
    "\n",
    "print(f\"Real-time processing complete\")\n",
    "print(f\"Extracted {len(security_entities)} security entities\")\n",
    "print(f\"Extracted {len(stream_relationships)} relationships\")\n",
    "print(f\"Detected {len(detected_events)} events\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Build Temporal Knowledge Graph\n",
    "\n",
    "Build and continuously update temporal knowledge graph from streaming data.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "builder = GraphBuilder()\n",
    "temporal_pattern_detector = TemporalPatternDetector()\n",
    "temporal_query = TemporalGraphQuery()\n",
    "graph_analyzer = GraphAnalyzer()\n",
    "\n",
    "# Build temporal KG from streaming data\n",
    "temporal_kg = builder.build(security_entities, stream_relationships)\n",
    "\n",
    "# Analyze graph structure in real-time\n",
    "metrics = graph_analyzer.compute_metrics(temporal_kg)\n",
    "centrality_calculator = CentralityCalculator()\n",
    "community_detector = CommunityDetector()\n",
    "connectivity_analyzer = ConnectivityAnalyzer()\n",
    "\n",
    "centrality_result = centrality_calculator.calculate_degree_centrality(temporal_kg)\n",
    "centrality_scores = centrality_result.get('centrality', {})\n",
    "communities = community_detector.detect_communities(temporal_kg)\n",
    "connectivity = connectivity_analyzer.analyze_connectivity(temporal_kg)\n",
    "\n",
    "print(f\"Built temporal knowledge graph from stream\")\n",
    "print(f\"  Entities: {len(temporal_kg.get('entities', []))}\")\n",
    "print(f\"  Relationships: {len(temporal_kg.get('relationships', []))}\")\n",
    "print(f\"  Graph density: {metrics.get('density', 0):.3f}\")\n",
    "print(f\"  Communities: {len(communities)}\")\n",
    "print(f\"  Central entities: {len([e for e, score in centrality_scores.items() if score > 0])}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 4: Real-Time Pattern Detection\n",
    "\n",
    "Detect temporal patterns and anomalies in real-time.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Detect temporal patterns\n",
    "temporal_patterns = temporal_pattern_detector.detect_temporal_patterns(\n",
    "    temporal_kg,\n",
    "    pattern_type=\"anomaly\",\n",
    "    min_frequency=1\n",
    ")\n",
    "\n",
    "# Real-time anomaly detection using inference\n",
    "inference_engine = InferenceEngine()\n",
    "rule_manager = RuleManager()\n",
    "explanation_generator = ExplanationGenerator()\n",
    "\n",
    "# Define real-time anomaly detection rules\n",
    "inference_engine.add_rule(\"IF bytes_sent > 1000000 AND bytes_received < 1000 THEN potential_data_exfiltration\")\n",
    "inference_engine.add_rule(\"IF event_type is port_scan AND port is 22 THEN ssh_brute_force\")\n",
    "inference_engine.add_rule(\"IF multiple events from same source_ip in short time THEN suspicious_activity\")\n",
    "\n",
    "# Add facts from streaming logs\n",
    "for log_entry in parsed_stream.data if parsed_stream and parsed_stream.data else []:\n",
    "    if isinstance(log_entry, dict):\n",
    "        inference_engine.add_fact({\n",
    "            \"source_ip\": log_entry.get(\"source_ip\", \"\"),\n",
    "            \"event_type\": log_entry.get(\"event_type\", \"\"),\n",
    "            \"bytes_sent\": log_entry.get(\"bytes_sent\", 0),\n",
    "            \"bytes_received\": log_entry.get(\"bytes_received\", 0),\n",
    "            \"port\": log_entry.get(\"port\", 0),\n",
    "            \"timestamp\": log_entry.get(\"timestamp\", \"\")\n",
    "        })\n",
    "\n",
    "inferred_anomalies = inference_engine.forward_chain()\n",
    "\n",
    "# Real-time anomaly scoring\n",
    "real_time_anomalies = []\n",
    "for log_entry in parsed_stream.data if parsed_stream and parsed_stream.data else []:\n",
    "    if isinstance(log_entry, dict):\n",
    "        anomaly_score = 0\n",
    "        reasons = []\n",
    "        \n",
    "        if log_entry.get(\"bytes_sent\", 0) > 1000000:\n",
    "            anomaly_score += 5\n",
    "            reasons.append(\"Unusually large data transfer\")\n",
    "        \n",
    "        if log_entry.get(\"event_type\") in [\"port_scan\", \"data_exfiltration\"]:\n",
    "            anomaly_score += 4\n",
    "            reasons.append(\"High-risk event type\")\n",
    "        \n",
    "        if log_entry.get(\"bytes_sent\", 0) > log_entry.get(\"bytes_received\", 0) * 100:\n",
    "            anomaly_score += 3\n",
    "            reasons.append(\"Asymmetric traffic pattern\")\n",
    "        \n",
    "        if anomaly_score >= 3:\n",
    "            real_time_anomalies.append({\n",
    "                \"source_ip\": log_entry.get(\"source_ip\", \"\"),\n",
    "                \"destination_ip\": log_entry.get(\"destination_ip\", \"\"),\n",
    "                \"event_type\": log_entry.get(\"event_type\", \"\"),\n",
    "                \"severity\": \"high\" if anomaly_score >= 5 else \"medium\",\n",
    "                \"score\": anomaly_score,\n",
    "                \"reasons\": reasons,\n",
    "                \"timestamp\": log_entry.get(\"timestamp\", \"\")\n",
    "            })\n",
    "\n",
    "print(f\"Detected {len(temporal_patterns)} temporal patterns\")\n",
    "print(f\"Inferred {len(inferred_anomalies)} anomalies from rules\")\n",
    "print(f\"Identified {len(real_time_anomalies)} real-time anomalies\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 5: Generate Real-Time Alerts\n",
    "\n",
    "Generate and send alerts for detected anomalies.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "quality_assessor = KGQualityAssessor()\n",
    "json_exporter = JSONExporter()\n",
    "csv_exporter = CSVExporter()\n",
    "report_generator = ReportGenerator()\n",
    "\n",
    "quality_score = quality_assessor.assess_overall_quality(temporal_kg)\n",
    "\n",
    "# Generate alerts\n",
    "alerts = []\n",
    "for anomaly in real_time_anomalies:\n",
    "    alert = {\n",
    "        \"alert_id\": f\"alert_{anomaly['source_ip']}_{int(time.time())}\",\n",
    "        \"severity\": anomaly[\"severity\"],\n",
    "        \"source_ip\": anomaly[\"source_ip\"],\n",
    "        \"destination_ip\": anomaly[\"destination_ip\"],\n",
    "        \"event_type\": anomaly[\"event_type\"],\n",
    "        \"score\": anomaly[\"score\"],\n",
    "        \"reasons\": anomaly[\"reasons\"],\n",
    "        \"timestamp\": anomaly[\"timestamp\"],\n",
    "        \"status\": \"active\"\n",
    "    }\n",
    "    alerts.append(alert)\n",
    "\n",
    "# Export alerts\n",
    "json_exporter.export_knowledge_graph(temporal_kg, os.path.join(temp_dir, \"realtime_kg.json\"))\n",
    "csv_exporter.export_entities(security_entities, os.path.join(temp_dir, \"realtime_entities.csv\"))\n",
    "\n",
    "report_data = {\n",
    "    \"summary\": f\"Real-time anomaly detection identified {len(real_time_anomalies)} anomalies\",\n",
    "    \"total_events\": len(parsed_stream.data) if parsed_stream and parsed_stream.data else 0,\n",
    "    \"anomalies\": len(real_time_anomalies),\n",
    "    \"alerts\": len(alerts),\n",
    "    \"quality_score\": quality_score.get('overall_score', 0),\n",
    "    \"high_severity\": len([a for a in alerts if a.get('severity') == 'high'])\n",
    "}\n",
    "\n",
    "report = report_generator.generate_report(report_data, format=\"markdown\")\n",
    "\n",
    "print(f\"Generated {len(alerts)} real-time alerts\")\n",
    "print(f\"High severity alerts: {len([a for a in alerts if a.get('severity') == 'high'])}\")\n",
    "print(f\"Report length: {len(report)} characters\")\n",
    "print(f\"Graph quality score: {quality_score.get('overall_score', 0):.3f}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 6: Real-Time Monitoring and Visualization\n",
    "\n",
    "Monitor security events in real-time and visualize results.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "kg_visualizer = KGVisualizer()\n",
    "temporal_visualizer = TemporalVisualizer()\n",
    "analytics_visualizer = AnalyticsVisualizer()\n",
    "\n",
    "kg_viz = kg_visualizer.visualize_network(temporal_kg, output=\"interactive\")\n",
    "temporal_viz = temporal_visualizer.visualize_timeline(temporal_kg, output=\"interactive\")\n",
    "analytics_viz = analytics_visualizer.visualize_analytics(temporal_kg, output=\"interactive\")\n",
    "\n",
    "print(f\"Real-time monitoring active\")\n",
    "print(f\"Monitoring {len(temporal_kg.get('entities', []))} entities in real-time\")\n",
    "print(f\"Active alerts: {len(alerts)}\")\n",
    "print(f\"Total modules used: 20+\")\n",
    "print(f\"Pipeline complete: Stream Logs → Real-Time Parse → Extract → Temporal KG → Pattern Detection → Anomaly Detection → Alerts → Monitor → Visualize\")\n"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
