{
 "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/01_Market_Data_Analysis.ipynb)\n",
    "\n",
    "# Market Data Analysis Pipeline\n",
    "\n",
    "## Overview\n",
    "\n",
    "This notebook demonstrates a complete market data analysis pipeline: stream market data from multiple sources (trading APIs, financial feeds, databases), build temporal market knowledge graph, analyze patterns, and predict trends.\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, CSVParser, StructuredDataParser\n",
    "- **Extraction**: NERExtractor, RelationExtractor, EventDetector, SemanticAnalyzer\n",
    "- **KG**: GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer\n",
    "- **Analytics**: CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n",
    "- **Reasoning**: InferenceEngine, RuleManager, ExplanationGenerator\n",
    "- **Quality**: KGQualityAssessor\n",
    "- **Export**: JSONExporter, CSVExporter, RDFExporter, ReportGenerator\n",
    "- **Visualization**: KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
    "\n",
    "### Pipeline\n",
    "\n",
    "**Stream Market Data → Parse → Extract Entities → Build Temporal Market KG → Analyze Patterns → Predict Trends → Generate Reports → Visualize**\n",
    "\n",
    "---\n",
    "\n",
    "## Step 1: Stream Market Data from Multiple Sources\n",
    "\n",
    "Stream market data from trading APIs, financial feeds, and databases.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from semantica.ingest import StreamIngestor, FileIngestor, WebIngestor, DBIngestor, FeedIngestor\n",
    "from semantica.parse import JSONParser, CSVParser, StructuredDataParser\n",
    "from semantica.semantic_extract import NERExtractor, RelationExtractor, EventDetector, SemanticAnalyzer\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, RDFExporter, ReportGenerator\n",
    "from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
    "import tempfile\n",
    "import os\n",
    "import json\n",
    "from datetime import datetime, timedelta\n",
    "\n",
    "stream_ingestor = StreamIngestor()\n",
    "file_ingestor = FileIngestor()\n",
    "web_ingestor = WebIngestor()\n",
    "db_ingestor = DBIngestor()\n",
    "feed_ingestor = FeedIngestor()\n",
    "\n",
    "json_parser = JSONParser()\n",
    "csv_parser = CSVParser()\n",
    "structured_parser = StructuredDataParser()\n",
    "\n",
    "# Real streaming sources for market data\n",
    "stream_sources = [\n",
    "    {\n",
    "        \"type\": \"kafka\",\n",
    "        \"topic\": \"market_data\",\n",
    "        \"bootstrap_servers\": [\"localhost:9092\"],\n",
    "        \"consumer_config\": {\"group_id\": \"market_analysis\"}\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",
    "    \"https://api.github.com/repos/ranaroussi/yfinance\"  # Yahoo Finance API\n",
    "]\n",
    "\n",
    "financial_feeds = [\n",
    "    \"https://feeds.reuters.com/reuters/businessNews\",\n",
    "    \"https://rss.cnn.com/rss/money_latest.rss\",\n",
    "    \"https://feeds.bloomberg.com/markets/news.rss\"\n",
    "]\n",
    "\n",
    "# Real database connection for market data\n",
    "db_connection_string = \"postgresql://user:password@localhost:5432/market_data_db\"\n",
    "db_query = \"SELECT symbol, price, volume, timestamp FROM market_data WHERE timestamp > NOW() - INTERVAL '1 hour' ORDER BY timestamp DESC LIMIT 1000\"\n",
    "\n",
    "temp_dir = tempfile.mkdtemp()\n",
    "\n",
    "# Sample streaming market data\n",
    "market_data_file = os.path.join(temp_dir, \"market_data_stream.json\")\n",
    "market_stream = [\n",
    "    {\n",
    "        \"symbol\": \"AAPL\",\n",
    "        \"price\": 175.50,\n",
    "        \"volume\": 1000000,\n",
    "        \"timestamp\": (datetime.now() - timedelta(minutes=5)).isoformat()\n",
    "    },\n",
    "    {\n",
    "        \"symbol\": \"MSFT\",\n",
    "        \"price\": 380.25,\n",
    "        \"volume\": 800000,\n",
    "        \"timestamp\": (datetime.now() - timedelta(minutes=4)).isoformat()\n",
    "    },\n",
    "    {\n",
    "        \"symbol\": \"GOOGL\",\n",
    "        \"price\": 142.80,\n",
    "        \"volume\": 1200000,\n",
    "        \"timestamp\": (datetime.now() - timedelta(minutes=3)).isoformat()\n",
    "    }\n",
    "]\n",
    "\n",
    "with open(market_data_file, 'w') as f:\n",
    "    json.dump(market_stream, f, indent=2)\n",
    "\n",
    "file_objects = file_ingestor.ingest_file(market_data_file, read_content=True)\n",
    "parsed_data = structured_parser.parse_data(market_data_file, data_format=\"json\")\n",
    "\n",
    "# Ingest from financial feeds\n",
    "financial_feed_list = []\n",
    "for feed_url in financial_feeds[:2]:\n",
    "    feed_data = feed_ingestor.ingest_feed(feed_url)\n",
    "    if feed_data:\n",
    "        financial_feed_list.append(feed_data)\n",
    "        print(f\"  Ingested feed: {feed_url}\")\n",
    "\n",
    "print(f\"\\n📊 Ingestion Summary:\")\n",
    "print(f\"  Market data files: {len([file_objects]) if file_objects else 0}\")\n",
    "print(f\"  Financial feeds: {len(financial_feed_list)}\")\n",
    "print(f\"  Streaming sources: {len(stream_sources)}\")\n",
    "print(f\"  Database sources: 1\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Extract Market Entities and Build Temporal Knowledge Graph\n",
    "\n",
    "Extract market entities and build temporal knowledge graph.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ner_extractor = NERExtractor()\n",
    "relation_extractor = RelationExtractor()\n",
    "event_detector = EventDetector()\n",
    "semantic_analyzer = SemanticAnalyzer()\n",
    "\n",
    "market_entities = []\n",
    "market_relationships = []\n",
    "\n",
    "# Extract from market data\n",
    "if parsed_data and parsed_data.get(\"data\"):\n",
    "    for market_entry in parsed_data.get(\"data\") if isinstance(parsed_data.get(\"data\"), list) else [parsed_data.get(\"data\")]:\n",
    "        if isinstance(market_entry, dict):\n",
    "            symbol = market_entry.get(\"symbol\", \"\")\n",
    "            \n",
    "            market_entities.append({\n",
    "                \"id\": symbol,\n",
    "                \"type\": \"Stock\",\n",
    "                \"name\": symbol,\n",
    "                \"properties\": {\n",
    "                    \"price\": market_entry.get(\"price\", 0),\n",
    "                    \"volume\": market_entry.get(\"volume\", 0),\n",
    "                    \"timestamp\": market_entry.get(\"timestamp\", \"\")\n",
    "                }\n",
    "            })\n",
    "            \n",
    "            # Price events\n",
    "            if market_entry.get(\"price\", 0) > 0:\n",
    "                market_relationships.append({\n",
    "                    \"source\": symbol,\n",
    "                    \"target\": f\"{symbol}_price_{market_entry.get('timestamp', '')}\",\n",
    "                    \"type\": \"has_price\",\n",
    "                    \"properties\": {\n",
    "                        \"price\": market_entry.get(\"price\", 0),\n",
    "                        \"timestamp\": market_entry.get(\"timestamp\", \"\")\n",
    "                    }\n",
    "                })\n",
    "\n",
    "builder = GraphBuilder()\n",
    "temporal_query = TemporalGraphQuery()\n",
    "temporal_pattern_detector = TemporalPatternDetector()\n",
    "graph_analyzer = GraphAnalyzer()\n",
    "\n",
    "market_kg = builder.build(market_entities, market_relationships)\n",
    "\n",
    "metrics = graph_analyzer.compute_metrics(market_kg)\n",
    "\n",
    "print(f\"Extracted {len(market_entities)} market entities\")\n",
    "print(f\"Extracted {len(market_relationships)} relationships\")\n",
    "print(f\"Built temporal market knowledge graph with {len(market_kg.get('entities', []))} entities\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Analyze Market Patterns\n",
    "\n",
    "Analyze market patterns using temporal analysis and graph analytics.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "centrality_calculator = CentralityCalculator()\n",
    "community_detector = CommunityDetector()\n",
    "connectivity_analyzer = ConnectivityAnalyzer()\n",
    "\n",
    "start_time = (datetime.now() - timedelta(hours=1)).isoformat()\n",
    "end_time = datetime.now().isoformat()\n",
    "\n",
    "temporal_results = temporal_query.query_time_range(\n",
    "    graph=market_kg,\n",
    "    query=\"Find market movements in the last hour\",\n",
    "    start_time=start_time,\n",
    "    end_time=end_time\n",
    ")\n",
    "\n",
    "temporal_patterns = temporal_pattern_detector.detect_temporal_patterns(\n",
    "    market_kg,\n",
    "    pattern_type=\"trend\",\n",
    "    min_frequency=1\n",
    ")\n",
    "\n",
    "centrality_result = centrality_calculator.calculate_degree_centrality(market_kg)\n",
    "centrality_scores = centrality_result.get('centrality', {})\n",
    "communities = community_detector.detect_communities(market_kg)\n",
    "connectivity = connectivity_analyzer.analyze_connectivity(market_kg)\n",
    "\n",
    "print(f\"Temporal query returned {len(temporal_results.get('entities', []))} entities\")\n",
    "print(f\"Detected {len(temporal_patterns)} temporal patterns\")\n",
    "print(f\"Central stocks: {len([e for e, score in centrality_scores.items() if score > 0])}\")\n",
    "print(f\"Communities: {len(communities)}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 4: Predict Market Trends\n",
    "\n",
    "Predict market trends using inference engine.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inference_engine = InferenceEngine()\n",
    "rule_manager = RuleManager()\n",
    "explanation_generator = ExplanationGenerator()\n",
    "\n",
    "# Market trend prediction rules\n",
    "inference_engine.add_rule(\"IF volume > 1000000 AND price_change > 0 THEN bullish_signal\")\n",
    "inference_engine.add_rule(\"IF volume > 1000000 AND price_change < 0 THEN bearish_signal\")\n",
    "inference_engine.add_rule(\"IF multiple stocks show same pattern THEN market_trend\")\n",
    "\n",
    "# Add facts from market data\n",
    "if parsed_data and parsed_data.get(\"data\"):\n",
    "    for market_entry in parsed_data.get(\"data\") if isinstance(parsed_data.get(\"data\"), list) else [parsed_data.get(\"data\")]:\n",
    "        if isinstance(market_entry, dict):\n",
    "            inference_engine.add_fact({\n",
    "                \"symbol\": market_entry.get(\"symbol\", \"\"),\n",
    "                \"volume\": market_entry.get(\"volume\", 0),\n",
    "                \"price\": market_entry.get(\"price\", 0)\n",
    "            })\n",
    "\n",
    "trend_predictions = inference_engine.forward_chain()\n",
    "\n",
    "print(f\"Generated {len(trend_predictions)} market trend predictions\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 5: Generate Reports and Visualize\n",
    "\n",
    "Generate market analysis 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",
    "rdf_exporter = RDFExporter()\n",
    "report_generator = ReportGenerator()\n",
    "\n",
    "quality_score = quality_assessor.assess_overall_quality(market_kg)\n",
    "\n",
    "json_exporter.export_knowledge_graph(market_kg, os.path.join(temp_dir, \"market_kg.json\"))\n",
    "csv_exporter.export_entities(market_entities, os.path.join(temp_dir, \"market_entities.csv\"))\n",
    "rdf_exporter.export_knowledge_graph(market_kg, os.path.join(temp_dir, \"market_kg.rdf\"))\n",
    "\n",
    "report_data = {\n",
    "    \"summary\": f\"Market data analysis identified {len(trend_predictions)} trend predictions from {len(market_entities)} entities\",\n",
    "    \"stocks_analyzed\": len(market_entities),\n",
    "    \"patterns\": len(temporal_patterns),\n",
    "    \"predictions\": len(trend_predictions),\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(market_kg, output=\"interactive\")\n",
    "temporal_viz = temporal_visualizer.visualize_timeline(market_kg, output=\"interactive\")\n",
    "analytics_viz = analytics_visualizer.visualize_analytics(market_kg, output=\"interactive\")\n",
    "\n",
    "print(f\"Total modules used: 20+\")\n",
    "print(f\"Pipeline complete: Stream Market Data → Parse → Extract → Build Temporal KG → Analyze Patterns → Predict Trends → Reports → Visualize\")\n"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
