Merge master branch into main with detailed README

Author: Drago-03

.bolt/config.json

@@ -0,0 +1,3 @@
+{
+  "template": "bolt-vite-react-ts"
+}

.bolt/prompt

@@ -0,0 +1,8 @@
+For all designs I ask you to make, have them be beautiful, not cookie cutter. Make webpages that are fully featured and worthy for production.
+
+By default, this template supports JSX syntax with Tailwind CSS classes, React hooks, and Lucide React for icons. Do not install other packages for UI themes, icons, etc unless absolutely necessary or I request them.
+
+Use icons from lucide-react for logos.
+
+Use stock photos from unsplash where appropriate, only valid URLs you know exist. Do not download the images, only link to them in image tags.
+

.env

@@ -0,0 +1,3 @@
+
+VITE_SUPABASE_ANON_KEY=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6Imhrc29jcXBvYWd0aGR3YWFpaGpmIiwicm9sZSI6ImFub24iLCJpYXQiOjE3NDEzNTgxNDgsImV4cCI6MjA1NjkzNDE0OH0.vo2irqrtwjbt66mBca9QUJFXVvLmCecgj1_9CgNTxiQ
+VITE_SUPABASE_URL=https://hksocqpoagthdwaaihjf.supabase.co

.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

README.md

@@ -1 +1,53 @@
-# SafalMudra
+# Financial Insights RAG Application
+
+This application provides real-time financial insights using Retrieval Augmented Generation (RAG) with Pathway, FAISS, and Ollama.
+
+## Prerequisites
+
+1. Install Ollama and download the Mistral model:
+```bash
+# macOS/Linux
+curl https://ollama.ai/install.sh | sh
+# Then run:
+ollama pull mistral
+```
+
+2. Install Python dependencies:
+```bash
+pip install -r requirements.txt
+```
+
+3. Start the backend server:
+```bash
+cd backend
+python main.py
+```
+
+4. Start the frontend development server:
+```bash
+npm run dev
+```
+
+## Features
+
+- Real-time financial data processing with Pathway
+- Vector similarity search using FAISS
+- Local LLM integration with Ollama (Mistral)
+- User authentication with Supabase
+- Query history tracking
+- Beautiful, responsive UI
+
+## Architecture
+
+- Frontend: React + TypeScript + Vite
+- Backend: FastAPI + Pathway
+- Vector Store: FAISS
+- LLM: Ollama (Mistral)
+- Authentication: Supabase
+- Real-time Processing: Pathway
+
+## Development
+
+1. The backend server runs on `http://localhost:8000`
+2. The frontend development server runs on `http://localhost:5173`
+3. Ollama serves the LLM on `http://localhost:11434`

backend/alert_manager.py

@@ -0,0 +1,48 @@
+from typing import List, Dict, Any
+import asyncio
+import json
+import websockets
+from datetime import datetime
+
+class AlertManager:
+    def __init__(self):
+        self.websocket_connections: List[websockets.WebSocketServerProtocol] = []
+        self.alert_history: List[Dict[str, Any]] = []
+        
+    async def start_websocket_server(self):
+        async def handler(websocket):
+            self.websocket_connections.append(websocket)
+            try:
+                async for message in websocket:
+                    # Handle incoming messages if needed
+                    pass
+            finally:
+                self.websocket_connections.remove(websocket)
+                
+        async with websockets.serve(handler, "localhost", 8765):
+            await asyncio.Future()  # run forever
+            
+    async def send_alert(self, alert_type: str, message: str, data: Dict[str, Any]):
+        alert = {
+            "type": alert_type,
+            "message": message,
+            "data": data,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        
+        self.alert_history.append(alert)
+        
+        # Send to all connected clients
+        websockets_to_remove = []
+        for ws in self.websocket_connections:
+            try:
+                await ws.send(json.dumps(alert))
+            except websockets.exceptions.ConnectionClosed:
+                websockets_to_remove.append(ws)
+                
+        # Clean up closed connections
+        for ws in websockets_to_remove:
+            self.websocket_connections.remove(ws)
+            
+    def get_alert_history(self) -> List[Dict[str, Any]]:
+        return self.alert_history

backend/data/sample.jsonl

@@ -0,0 +1,3 @@
+{"content": "The Federal Reserve maintains its commitment to a 2% inflation target over the longer run and considers this level consistent with its mandate of maximum employment and price stability.", "source": "Federal Reserve Statement", "timestamp": "2024-03-01T10:00:00Z"}
+{"content": "SEC regulations require public companies to file annual reports (Form 10-K) and quarterly reports (Form 10-Q) to provide regular updates on their financial position and operations.", "source": "SEC Guidelines", "timestamp": "2024-03-01T10:01:00Z"}
+{"content": "The Basel III framework requires banks to maintain a minimum total capital ratio of 8% of risk-weighted assets, with additional buffers for systemically important financial institutions.", "source": "Basel Committee Guidelines", "timestamp": "2024-03-01T10:02:00Z"}

backend/embeddings.py

@@ -0,0 +1,48 @@
+from sentence_transformers import SentenceTransformer
+import numpy as np
+import faiss
+import pickle
+from typing import List, Tuple
+import os
+
+class EmbeddingStore:
+    def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
+        self.model = SentenceTransformer(model_name)
+        self.dimension = 384  # Dimension for all-MiniLM-L6-v2
+        self.index = faiss.IndexFlatL2(self.dimension)
+        self.texts: List[str] = []
+        self.sources: List[str] = []
+
+    def add_texts(self, texts: List[str], sources: List[str]) -> None:
+        if len(texts) != len(sources):
+            raise ValueError("Number of texts and sources must match")
+        
+        embeddings = self.model.encode(texts)
+        self.index.add(np.array(embeddings).astype('float32'))
+        self.texts.extend(texts)
+        self.sources.extend(sources)
+
+    def similarity_search(self, query: str, k: int = 3) -> List[Tuple[str, str, float]]:
+        query_embedding = self.model.encode([query])
+        distances, indices = self.index.search(np.array(query_embedding).astype('float32'), k)
+        
+        results = []
+        for i, idx in enumerate(indices[0]):
+            if idx < len(self.texts):  # Ensure index is valid
+                results.append((
+                    self.texts[idx],
+                    self.sources[idx],
+                    float(distances[0][i])
+                ))
+        return results
+
+    def save(self, directory: str) -> None:
+        os.makedirs(directory, exist_ok=True)
+        faiss.write_index(self.index, os.path.join(directory, "faiss_index"))
+        with open(os.path.join(directory, "metadata.pkl"), "wb") as f:
+            pickle.dump((self.texts, self.sources), f)
+
+    def load(self, directory: str) -> None:
+        self.index = faiss.read_index(os.path.join(directory, "faiss_index"))
+        with open(os.path.join(directory, "metadata.pkl"), "rb") as f:
+            self.texts, self.sources = pickle.load(f)

backend/llm.py

@@ -0,0 +1,38 @@
+import requests
+from typing import List, Dict, Any
+
+class LLM:
+    def __init__(self, model: str = "mistral"):
+        self.model = model
+        self.api_url = "http://localhost:11434/api/generate"
+        
+    async def generate_response(self, query: str, context: List[Dict[str, Any]]) -> str:
+        # Format context for the prompt
+        context_str = "\n".join([
+            f"Source ({item['source']}): {item['text']}"
+            for item in context
+        ])
+        
+        prompt = f"""You are a financial analysis AI assistant. Use the following context to answer the question. 
+If you cannot answer the question based on the context, say so.
+
+Context:
+{context_str}
+
+Question: {query}
+
+Answer:"""
+
+        response = requests.post(
+            self.api_url,
+            json={
+                "model": self.model,
+                "prompt": prompt,
+                "stream": False
+            }
+        )
+        
+        if response.status_code != 200:
+            raise Exception("Failed to get response from LLM")
+            
+        return response.json()["response"]

backend/main.py

@@ -0,0 +1,131 @@
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+import pathway as pw
+import asyncio
+import uvicorn
+
+from embeddings import EmbeddingStore
+from llm import LLM
+from metrics_extractor import MetricsExtractor
+from alert_manager import AlertManager
+
+app = FastAPI()
+
+# Configure CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["http://localhost:5173"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Initialize components
+embedding_store = EmbeddingStore()
+llm = LLM()
+metrics_extractor = MetricsExtractor()
+alert_manager = AlertManager()
+
+class QueryRequest(BaseModel):
+    query: str
+    filters: Optional[dict] = None
+
+class QueryResponse(BaseModel):
+    answer: str
+    context: List[str]
+    sources: List[str]
+    timestamp: str
+    metrics: Optional[Dict[str, Any]] = None
+
+@app.post("/query")
+async def query(request: QueryRequest) -> QueryResponse:
+    try:
+        # Get relevant documents from FAISS
+        results = embedding_store.similarity_search(request.query)
+        
+        # Format context for LLM
+        context = [
+            {"text": text, "source": source}
+            for text, source, _ in results
+        ]
+        
+        # Generate response using LLM
+        answer = await llm.generate_response(request.query, context)
+        
+        # Extract financial metrics if present
+        metrics = None
+        if any(keyword in request.query.lower() for keyword in ['metrics', 'financial', 'numbers']):
+            metrics_data = {}
+            for text, _, _ in results:
+                extracted = metrics_extractor.extract_metrics(text)
+                for field, value in extracted.__dict__.items():
+                    if value is not None:
+                        metrics_data[field] = value
+            if metrics_data:
+                metrics = metrics_data
+                
+                # Send alert if significant metrics are found
+                await alert_manager.send_alert(
+                    "metrics_found",
+                    "New financial metrics detected in query",
+                    metrics_data
+                )
+        
+        return QueryResponse(
+            answer=answer,
+            context=[text for text, _, _ in results],
+            sources=[source for _, source, _ in results],
+            timestamp=datetime.utcnow().isoformat(),
+            metrics=metrics
+        )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/alerts")
+async def get_alerts():
+    return alert_manager.get_alert_history()
+
+# Pathway data processing pipeline
+def setup_pathway_pipeline():
+    with pw.Config.interactive():
+        # Input table for financial data
+        input_table = pw.io.fs.read(
+            "data/*.jsonl",
+            mode="streaming",
+            format="json"
+        )
+        
+        # Process and index new documents
+        def process_document(doc):
+            embedding_store.add_texts(
+                [doc["content"]],
+                [doc["source"]]
+            )
+            
+            # Extract and alert on new financial metrics
+            metrics = metrics_extractor.extract_metrics(doc["content"])
+            if any(value is not None for value in metrics.__dict__.values()):
+                asyncio.create_task(
+                    alert_manager.send_alert(
+                        "new_document_metrics",
+                        f"New financial metrics found in document from {doc['source']}",
+                        metrics.__dict__
+                    )
+                )
+            return doc
+        
+        processed = input_table.select(process_document)
+        processed.run()
+
+if __name__ == "__main__":
+    # Start WebSocket server for alerts
+    asyncio.create_task(alert_manager.start_websocket_server())
+    
+    # Start Pathway pipeline
+    asyncio.create_task(setup_pathway_pipeline())
+    
+    # Run FastAPI server
+    uvicorn.run(app, host="0.0.0.0", port=8000)