{
 "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/trading/04_Real_Time_Monitoring.ipynb)\n",
    "\n",
    "# Real-Time Trading Monitoring Pipeline\n",
    "\n",
    "## Overview\n",
    "\n",
    "This notebook demonstrates a complete real-time trading monitoring pipeline: stream trading data from multiple sources (trading platforms, market data streams, databases), build temporal knowledge graph, monitor positions in real-time, detect anomalies, and generate alerts.\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\n",
    "- **Extraction**: NERExtractor, RelationExtractor, EventDetector, TripleExtractor\n",
    "- **KG**: GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, 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",
    "**Real-Time Trading Streams → Parse → Extract Entities → Build Temporal KG → Monitor Positions → Detect Anomalies → Generate Alerts → Visualize**\n",
    "\n",
    "---\n",
    "\n",
    "## Step 1: Stream Trading Data from Multiple Sources\n",
    "\n",
    "Stream trading data from trading platforms, market data streams, and databases.\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\n",
    "from semantica.semantic_extract import NERExtractor, RelationExtractor, EventDetector, TripleExtractor\n",
    "from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, 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",
    "json_parser = JSONParser()\n",
    "structured_parser = StructuredDataParser()\n",
    "\n",
    "# Real streaming sources for trading data\n",
    "stream_sources = [\n",
    "    {\n",
    "        \"type\": \"kafka\",\n",
    "        \"topic\": \"trading_data\",\n",
    "        \"bootstrap_servers\": [\"localhost:9092\"],\n",
    "        \"consumer_config\": {\"group_id\": \"trading_monitor\"}\n",
    "    },\n",
    "    {\n",
    "        \"type\": \"rabbitmq\",\n",
    "        \"queue\": \"trading_events\",\n",
    "        \"connection_url\": \"amqp://user:password@localhost:5672/\"\n",
    "    }\n",
    "]\n",
    "\n",
    "# Real market data APIs\n",
    "market_apis = [\n",
    "    \"https://api.polygon.io/v2/aggs/ticker/AAPL/range/1/minute/2024-01-15/2024-01-15\",  # Polygon.io\n",
    "    \"https://www.alphavantage.co/query?function=TIME_SERIES_INTRADAY&symbol=AAPL&interval=1min&apikey=demo\"  # Alpha Vantage\n",
    "]\n",
    "\n",
    "# Real database connection for trading positions\n",
    "db_connection_string = \"postgresql://user:password@localhost:5432/trading_db\"\n",
    "db_query = \"SELECT position_id, symbol, quantity, entry_price, current_price, timestamp FROM positions WHERE timestamp > NOW() - INTERVAL '1 hour' ORDER BY timestamp DESC\"\n",
    "\n",
    "temp_dir = tempfile.mkdtemp()\n",
    "\n",
    "# Sample real-time trading data\n",
    "trading_stream_file = os.path.join(temp_dir, \"trading_stream.json\")\n",
    "trading_stream = [\n",
    "    {\n",
    "        \"position_id\": \"POS-001\",\n",
    "        \"symbol\": \"AAPL\",\n",
    "        \"quantity\": 100,\n",
    "        \"entry_price\": 175.00,\n",
    "        \"current_price\": 175.50,\n",
    "        \"timestamp\": (datetime.now() - timedelta(minutes=5)).isoformat()\n",
    "    },\n",
    "    {\n",
    "        \"position_id\": \"POS-002\",\n",
    "        \"symbol\": \"MSFT\",\n",
    "        \"quantity\": 50,\n",
    "        \"entry_price\": 380.00,\n",
    "        \"current_price\": 380.25,\n",
    "        \"timestamp\": (datetime.now() - timedelta(minutes=4)).isoformat()\n",
    "    },\n",
    "    {\n",
    "        \"position_id\": \"POS-003\",\n",
    "        \"symbol\": \"GOOGL\",\n",
    "        \"quantity\": 75,\n",
    "        \"entry_price\": 142.00,\n",
    "        \"current_price\": 142.80,\n",
    "        \"timestamp\": (datetime.now() - timedelta(minutes=3)).isoformat()\n",
    "    }\n",
    "]\n",
    "\n",
    "with open(trading_stream_file, 'w') as f:\n",
    "    json.dump(trading_stream, f, indent=2)\n",
    "\n",
    "file_objects = file_ingestor.ingest_file(trading_stream_file, read_content=True)\n",
    "parsed_data = structured_parser.parse_data(trading_stream_file, data_format=\"json\")\n",
    "\n",
    "print(f\"\\n📊 Ingestion Summary:\")\n",
    "print(f\"  Trading stream files: {len([file_objects]) if file_objects else 0}\")\n",
    "print(f\"  Streaming sources: {len(stream_sources)}\")\n",
    "print(f\"  Database sources: 1\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Extract Trading Entities and Build Temporal Knowledge Graph\n",
    "\n",
    "Extract trading entities and build temporal knowledge graph for real-time monitoring.\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",
    "trading_entities = []\n",
    "trading_relationships = []\n",
    "\n",
    "# Extract from trading stream data\n",
    "if parsed_data and parsed_data.get(\"data\"):\n",
    "    for position in parsed_data.get(\"data\") if isinstance(parsed_data.get(\"data\"), list) else [parsed_data.get(\"data\")]:\n",
    "        if isinstance(position, dict):\n",
    "            position_id = position.get(\"position_id\", \"\")\n",
    "            symbol = position.get(\"symbol\", \"\")\n",
    "            \n",
    "            trading_entities.append({\n",
    "                \"id\": position_id,\n",
    "                \"type\": \"Position\",\n",
    "                \"name\": position_id,\n",
    "                \"properties\": {\n",
    "                    \"quantity\": position.get(\"quantity\", 0),\n",
    "                    \"entry_price\": position.get(\"entry_price\", 0),\n",
    "                    \"current_price\": position.get(\"current_price\", 0),\n",
    "                    \"timestamp\": position.get(\"timestamp\", \"\")\n",
    "                }\n",
    "            })\n",
    "            \n",
    "            trading_entities.append({\n",
    "                \"id\": symbol,\n",
    "                \"type\": \"Stock\",\n",
    "                \"name\": symbol,\n",
    "                \"properties\": {}\n",
    "            })\n",
    "            \n",
    "            trading_relationships.append({\n",
    "                \"source\": position_id,\n",
    "                \"target\": symbol,\n",
    "                \"type\": \"holds\",\n",
    "                \"properties\": {\n",
    "                    \"quantity\": position.get(\"quantity\", 0),\n",
    "                    \"timestamp\": position.get(\"timestamp\", \"\")\n",
    "                }\n",
    "            })\n",
    "\n",
    "builder = GraphBuilder()\n",
    "temporal_query = TemporalGraphQuery()\n",
    "temporal_pattern_detector = TemporalPatternDetector()\n",
    "graph_analyzer = GraphAnalyzer()\n",
    "\n",
    "trading_kg = builder.build(trading_entities, trading_relationships)\n",
    "\n",
    "metrics = graph_analyzer.compute_metrics(trading_kg)\n",
    "\n",
    "print(f\"Extracted {len(trading_entities)} trading entities\")\n",
    "print(f\"Extracted {len(trading_relationships)} relationships\")\n",
    "print(f\"Built temporal trading knowledge graph with {len(trading_kg.get('entities', []))} entities\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Monitor Positions in Real-Time\n",
    "\n",
    "Monitor trading positions using temporal queries.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "centrality_calculator = CentralityCalculator()\n",
    "community_detector = CommunityDetector()\n",
    "connectivity_analyzer = ConnectivityAnalyzer()\n",
    "\n",
    "current_time = datetime.now().isoformat()\n",
    "start_time = (datetime.now() - timedelta(minutes=10)).isoformat()\n",
    "\n",
    "# Query current positions\n",
    "current_positions = temporal_query.query_time_range(\n",
    "    graph=trading_kg,\n",
    "    query=\"Find current positions\",\n",
    "    start_time=start_time,\n",
    "    end_time=current_time\n",
    ")\n",
    "\n",
    "# Monitor position changes\n",
    "position_changes = []\n",
    "if parsed_data and parsed_data.get(\"data\"):\n",
    "    for position in parsed_data.get(\"data\") if isinstance(parsed_data.get(\"data\"), list) else [parsed_data.get(\"data\")]:\n",
    "        if isinstance(position, dict):\n",
    "            price_change = position.get(\"current_price\", 0) - position.get(\"entry_price\", 0)\n",
    "            price_change_percent = (price_change / position.get(\"entry_price\", 1)) * 100 if position.get(\"entry_price\", 0) > 0 else 0\n",
    "            \n",
    "            position_changes.append({\n",
    "                \"position_id\": position.get(\"position_id\", \"\"),\n",
    "                \"symbol\": position.get(\"symbol\", \"\"),\n",
    "                \"price_change\": price_change,\n",
    "                \"price_change_percent\": price_change_percent,\n",
    "                \"timestamp\": position.get(\"timestamp\", \"\")\n",
    "            })\n",
    "\n",
    "centrality_result = centrality_calculator.calculate_degree_centrality(trading_kg)\n",
    "centrality_scores = centrality_result.get('centrality', {})\n",
    "communities = community_detector.detect_communities(trading_kg)\n",
    "connectivity = connectivity_analyzer.analyze_connectivity(trading_kg)\n",
    "\n",
    "print(f\"Monitoring {len(current_positions.get('entities', []))} current positions\")\n",
    "print(f\"Position changes tracked: {len(position_changes)}\")\n",
    "print(f\"Central positions: {len([e for e, score in centrality_scores.items() if score > 0])}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 4: Detect Anomalies and Generate Alerts\n",
    "\n",
    "Detect anomalies in trading positions and generate alerts.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inference_engine = InferenceEngine()\n",
    "rule_manager = RuleManager()\n",
    "explanation_generator = ExplanationGenerator()\n",
    "\n",
    "# Anomaly detection rules\n",
    "inference_engine.add_rule(\"IF price_change_percent > 5 AND quantity > 100 THEN large_gain_alert\")\n",
    "inference_engine.add_rule(\"IF price_change_percent < -5 AND quantity > 100 THEN large_loss_alert\")\n",
    "inference_engine.add_rule(\"IF price_change_percent > 10 THEN extreme_movement_alert\")\n",
    "\n",
    "# Detect anomalies\n",
    "anomalies = []\n",
    "alerts = []\n",
    "\n",
    "for position_change in position_changes:\n",
    "    anomaly_score = 0\n",
    "    reasons = []\n",
    "    \n",
    "    if abs(position_change.get(\"price_change_percent\", 0)) > 5:\n",
    "        anomaly_score += 3\n",
    "        reasons.append(\"Significant price movement\")\n",
    "    \n",
    "    if position_change.get(\"price_change_percent\", 0) > 10:\n",
    "        anomaly_score += 5\n",
    "        reasons.append(\"Extreme price movement\")\n",
    "    \n",
    "    if anomaly_score >= 3:\n",
    "        anomaly = {\n",
    "            \"position_id\": position_change.get(\"position_id\", \"\"),\n",
    "            \"symbol\": position_change.get(\"symbol\", \"\"),\n",
    "            \"severity\": \"high\" if anomaly_score >= 5 else \"medium\",\n",
    "            \"score\": anomaly_score,\n",
    "            \"reasons\": reasons,\n",
    "            \"timestamp\": position_change.get(\"timestamp\", \"\")\n",
    "        }\n",
    "        anomalies.append(anomaly)\n",
    "        \n",
    "        alert = {\n",
    "            \"alert_id\": f\"alert_{position_change.get('position_id', '')}_{int(time.time())}\",\n",
    "            \"type\": \"trading_anomaly\",\n",
    "            \"severity\": anomaly[\"severity\"],\n",
    "            \"position\": position_change.get(\"position_id\", \"\"),\n",
    "            \"symbol\": position_change.get(\"symbol\", \"\"),\n",
    "            \"message\": f\"Anomaly detected: {', '.join(reasons)}\",\n",
    "            \"timestamp\": position_change.get(\"timestamp\", \"\")\n",
    "        }\n",
    "        alerts.append(alert)\n",
    "        \n",
    "        inference_engine.add_fact({\n",
    "            \"position_id\": position_change.get(\"position_id\", \"\"),\n",
    "            \"price_change_percent\": position_change.get(\"price_change_percent\", 0),\n",
    "            \"quantity\": position_change.get(\"quantity\", 0) if \"quantity\" in position_change else 0\n",
    "        })\n",
    "\n",
    "detected_anomalies = inference_engine.forward_chain()\n",
    "\n",
    "print(f\"Detected {len(anomalies)} trading anomalies\")\n",
    "print(f\"Generated {len(alerts)} alerts\")\n",
    "print(f\"Inferred {len(detected_anomalies)} anomaly patterns\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 5: Generate Reports and Visualize\n",
    "\n",
    "Generate real-time monitoring reports and visualize results.\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(trading_kg)\n",
    "\n",
    "json_exporter.export_knowledge_graph(trading_kg, os.path.join(temp_dir, \"trading_kg.json\"))\n",
    "csv_exporter.export_entities(trading_entities, os.path.join(temp_dir, \"trading_entities.csv\"))\n",
    "\n",
    "report_data = {\n",
    "    \"summary\": f\"Real-time monitoring detected {len(anomalies)} anomalies and generated {len(alerts)} alerts\",\n",
    "    \"positions_monitored\": len([e for e in trading_entities if e.get(\"type\") == \"Position\"]),\n",
    "    \"anomalies\": len(anomalies),\n",
    "    \"alerts\": len(alerts),\n",
    "    \"quality_score\": quality_score.get('overall_score', 0)\n",
    "}\n",
    "\n",
    "report = report_generator.generate_report(report_data, format=\"markdown\")\n",
    "\n",
    "kg_visualizer = KGVisualizer()\n",
    "temporal_visualizer = TemporalVisualizer()\n",
    "analytics_visualizer = AnalyticsVisualizer()\n",
    "\n",
    "kg_viz = kg_visualizer.visualize_network(trading_kg, output=\"interactive\")\n",
    "temporal_viz = temporal_visualizer.visualize_timeline(trading_kg, output=\"interactive\")\n",
    "analytics_viz = analytics_visualizer.visualize_analytics(trading_kg, output=\"interactive\")\n",
    "\n",
    "print(f\"Real-time monitoring active\")\n",
    "print(f\"Total modules used: 20+\")\n",
    "print(f\"Pipeline complete: Real-Time Streams → Parse → Extract → Build Temporal KG → Monitor Positions → Detect Anomalies → Alerts → Reports → Visualize\")\n"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
