implement NegaBot API with FastAPI for tweet sentiment classification and add SQLite logging system

api.py

@@ -0,0 +1,632 @@
+"""
+NegaBot API - FastAPI application for tweet sentiment classification
+"""
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import HTMLResponse, Response
+from pydantic import BaseModel, Field
+from typing import List, Optional
+import logging
+from datetime import datetime
+import json
+from model import get_model
+from database import log_prediction, get_all_predictions
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Initialize FastAPI app
+app = FastAPI(
+    title="NegaBot API",
+    description="Tweet Sentiment Classification API using NegaBot model",
+    version="1.0.0"
+)
+
+# Pydantic models for request/response validation
+class TweetRequest(BaseModel):
+    text: str = Field(..., min_length=1, max_length=1000, description="Tweet text to analyze")
+    metadata: Optional[dict] = Field(default=None, description="Optional metadata")
+
+class TweetResponse(BaseModel):
+    text: str
+    sentiment: str
+    confidence: float
+    predicted_class: int
+    probabilities: dict
+    timestamp: str
+    request_id: Optional[str] = None
+
+class BatchTweetRequest(BaseModel):
+    tweets: List[str] = Field(..., min_items=1, max_items=50, description="List of tweets to analyze")
+    metadata: Optional[dict] = Field(default=None, description="Optional metadata")
+
+class BatchTweetResponse(BaseModel):
+    results: List[TweetResponse]
+    total_processed: int
+    timestamp: str
+
+class HealthResponse(BaseModel):
+    status: str
+    model_loaded: bool
+    timestamp: str
+
+# Global variables
+model = None
+
+@app.on_event("startup")
+async def startup_event():
+    """Initialize the model on startup"""
+    global model
+    try:
+        logger.info("Starting NegaBot API...")
+        model = get_model()
+        logger.info("Model loaded successfully")
+    except Exception as e:
+        logger.error(f"Failed to load model: {str(e)}")
+        raise e
+
+@app.get("/", response_model=dict)
+async def root():
+    """Root endpoint with API information"""
+    return {
+        "message": "Welcome to NegaBot API",
+        "version": "1.0.0",
+        "description": "Tweet Sentiment Classification using NegaBot model",
+        "endpoints": {
+            "predict": "/predict - Single tweet prediction",
+            "batch_predict": "/batch_predict - Multiple tweets prediction",
+            "health": "/health - API health check",
+            "stats": "/stats - Prediction statistics",
+            "dashboard": "/dashboard - Interactive analytics dashboard",
+            "dashboard_data": "/dashboard/data - Dashboard data as JSON",
+            "download_csv": "/download/predictions.csv - Download predictions as CSV",
+            "download_json": "/download/predictions.json - Download predictions as JSON"
+        }
+    }
+
+@app.get("/health", response_model=HealthResponse)
+async def health_check():
+    """Health check endpoint"""
+    return HealthResponse(
+        status="healthy" if model is not None else "unhealthy",
+        model_loaded=model is not None,
+        timestamp=datetime.now().isoformat()
+    )
+
+@app.post("/predict", response_model=TweetResponse)
+async def predict_sentiment(request: TweetRequest):
+    """
+    Predict sentiment for a single tweet
+    
+    Args:
+        request: TweetRequest containing the tweet text
+        
+    Returns:
+        TweetResponse with prediction results
+    """
+    try:
+        if model is None:
+            raise HTTPException(status_code=503, detail="Model not loaded")
+        
+        # Get prediction from model
+        result = model.predict(request.text)
+        
+        # Create response
+        response = TweetResponse(
+            text=result["text"],
+            sentiment=result["sentiment"],
+            confidence=result["confidence"],
+            predicted_class=result["predicted_class"],
+            probabilities=result["probabilities"],
+            timestamp=datetime.now().isoformat()
+        )
+        
+        # Log the prediction
+        log_prediction(
+            text=request.text,
+            sentiment=result["sentiment"],
+            confidence=result["confidence"],
+            metadata=request.metadata
+        )
+        
+        logger.info(f"Prediction made: {result['sentiment']} (confidence: {result['confidence']:.2%})")
+        return response
+        
+    except Exception as e:
+        logger.error(f"Error in prediction: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Prediction failed: {str(e)}")
+
+@app.post("/batch_predict", response_model=BatchTweetResponse)
+async def batch_predict_sentiment(request: BatchTweetRequest):
+    """
+    Predict sentiment for multiple tweets
+    
+    Args:
+        request: BatchTweetRequest containing list of tweets
+        
+    Returns:
+        BatchTweetResponse with all prediction results
+    """
+    try:
+        if model is None:
+            raise HTTPException(status_code=503, detail="Model not loaded")
+        
+        # Get predictions for all tweets
+        results = model.batch_predict(request.tweets)
+        
+        # Create response objects
+        responses = []
+        for result in results:
+            response = TweetResponse(
+                text=result["text"],
+                sentiment=result["sentiment"],
+                confidence=result["confidence"],
+                predicted_class=result["predicted_class"],
+                probabilities=result["probabilities"],
+                timestamp=datetime.now().isoformat()
+            )
+            responses.append(response)
+            
+            # Log each prediction
+            log_prediction(
+                text=result["text"],
+                sentiment=result["sentiment"],
+                confidence=result["confidence"],
+                metadata=request.metadata
+            )
+        
+        batch_response = BatchTweetResponse(
+            results=responses,
+            total_processed=len(responses),
+            timestamp=datetime.now().isoformat()
+        )
+        
+        logger.info(f"Batch prediction completed: {len(responses)} tweets processed")
+        return batch_response
+        
+    except Exception as e:
+        logger.error(f"Error in batch prediction: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Batch prediction failed: {str(e)}")
+
+@app.get("/stats", response_model=dict)
+async def get_prediction_stats():
+    """
+    Get prediction statistics
+    
+    Returns:
+        Dictionary with prediction statistics
+    """
+    try:
+        predictions = get_all_predictions()
+        
+        if not predictions:
+            return {
+                "total_predictions": 0,
+                "positive_count": 0,
+                "negative_count": 0,
+                "average_confidence": 0,
+                "message": "No predictions found"
+            }
+        
+        total = len(predictions)
+        positive_count = sum(1 for p in predictions if p["sentiment"] == "Positive")
+        negative_count = total - positive_count
+        avg_confidence = sum(p["confidence"] for p in predictions) / total
+        
+        stats = {
+            "total_predictions": total,
+            "positive_count": positive_count,
+            "negative_count": negative_count,
+            "positive_percentage": round((positive_count / total) * 100, 2),
+            "negative_percentage": round((negative_count / total) * 100, 2),
+            "average_confidence": round(avg_confidence, 4),
+            "last_updated": datetime.now().isoformat()
+        }
+        
+        return stats
+        
+    except Exception as e:
+        logger.error(f"Error getting stats: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Failed to get statistics: {str(e)}")
+
+@app.get("/dashboard/data", response_model=dict)
+async def get_dashboard_data():
+    """
+    Get dashboard data as JSON for API consumption
+    """
+    try:
+        predictions = get_all_predictions()
+        
+        if not predictions:
+            return {
+                "metrics": {
+                    "total_predictions": 0,
+                    "positive_count": 0,
+                    "negative_count": 0,
+                    "average_confidence": 0
+                },
+                "recent_predictions": [],
+                "message": "No predictions found"
+            }
+        
+        # Calculate metrics
+        total = len(predictions)
+        positive_count = sum(1 for p in predictions if p["sentiment"] == "Positive")
+        negative_count = total - positive_count
+        avg_confidence = sum(p["confidence"] for p in predictions) / total
+        
+        # Get recent predictions (last 20)
+        recent_predictions = sorted(predictions, key=lambda x: x["created_at"], reverse=True)[:20]
+        
+        return {
+            "metrics": {
+                "total_predictions": total,
+                "positive_count": positive_count,
+                "negative_count": negative_count,
+                "positive_percentage": round((positive_count / total) * 100, 2),
+                "negative_percentage": round((negative_count / total) * 100, 2),
+                "average_confidence": round(avg_confidence, 4)
+            },
+            "recent_predictions": recent_predictions,
+            "last_updated": datetime.now().isoformat()
+        }
+        
+    except Exception as e:
+        logger.error(f"Error getting dashboard data: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Failed to get dashboard data: {str(e)}")
+
+@app.get("/download/predictions.csv")
+async def download_predictions_csv():
+    """
+    Download all predictions as CSV file
+    """
+    try:
+        predictions = get_all_predictions()
+        
+        if not predictions:
+            raise HTTPException(status_code=404, detail="No predictions found to download")
+        
+        # Convert to pandas DataFrame for easy CSV export
+        import pandas as pd
+        df = pd.DataFrame(predictions)
+        
+        # Convert to CSV
+        csv_content = df.to_csv(index=False)
+        
+        # Generate filename with timestamp
+        filename = f"negabot_predictions_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
+        
+        return Response(
+            content=csv_content,
+            media_type="text/csv",
+            headers={"Content-Disposition": f"attachment; filename={filename}"}
+        )
+        
+    except Exception as e:
+        logger.error(f"Error downloading CSV: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Failed to download CSV: {str(e)}")
+
+@app.get("/download/predictions.json")
+async def download_predictions_json():
+    """
+    Download all predictions as JSON file
+    """
+    try:
+        predictions = get_all_predictions()
+        
+        if not predictions:
+            raise HTTPException(status_code=404, detail="No predictions found to download")
+        
+        # Convert to JSON
+        json_content = json.dumps(predictions, indent=2, default=str)
+        
+        # Generate filename with timestamp
+        filename = f"negabot_predictions_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
+        
+        return Response(
+            content=json_content,
+            media_type="application/json",
+            headers={"Content-Disposition": f"attachment; filename={filename}"}
+        )
+        
+    except Exception as e:
+        logger.error(f"Error downloading JSON: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Failed to download JSON: {str(e)}")
+
+@app.get("/dashboard", response_class=HTMLResponse)
+async def dashboard():
+    """
+    Serve the analytics dashboard as HTML
+    """
+    try:
+        import pandas as pd
+        import plotly.express as px
+        import plotly.graph_objects as go
+        
+        # Get prediction data
+        predictions = get_all_predictions()
+        
+        if not predictions:
+            html_content = """
+            <!DOCTYPE html>
+            <html>
+            <head>
+                <title>NegaBot Dashboard</title>
+                <style>
+                    body { font-family: Arial, sans-serif; margin: 40px; }
+                    .container { max-width: 800px; margin: 0 auto; text-align: center; }
+                    .warning { background-color: #fff3cd; border: 1px solid #ffeaa7; padding: 20px; border-radius: 8px; }
+                </style>
+            </head>
+            <body>
+                <div class="container">
+                    <h1>🤖 NegaBot Analytics Dashboard</h1>
+                    <div class="warning">
+                        <h3>📭 No prediction data found</h3>
+                        <p>Make some predictions using the API first!</p>
+                        <p><strong>Quick Start:</strong></p>
+                        <ol>
+                            <li>Use POST to <code>/predict</code> endpoint</li>
+                            <li>Refresh this dashboard to see analytics</li>
+                        </ol>
+                        <p><strong>Available downloads:</strong></p>
+                        <p>
+                            <a href="/download/predictions.csv" style="color: #007bff; text-decoration: none;">📥 CSV Format</a> | 
+                            <a href="/download/predictions.json" style="color: #007bff; text-decoration: none;">📥 JSON Format</a>
+                        </p>
+                    </div>
+                </div>
+            </body>
+            </html>
+            """
+            return HTMLResponse(content=html_content)
+        
+        # Process data
+        df = pd.DataFrame(predictions)
+        df['created_at'] = pd.to_datetime(df['created_at'])
+        
+        # Calculate metrics
+        total_predictions = len(df)
+        positive_count = len(df[df['sentiment'] == 'Positive'])
+        negative_count = total_predictions - positive_count
+        avg_confidence = df['confidence'].mean()
+        
+        # Create sentiment distribution chart
+        sentiment_counts = df['sentiment'].value_counts()
+        fig_pie = px.pie(
+            values=sentiment_counts.values,
+            names=sentiment_counts.index,
+            title="Sentiment Distribution",
+            color_discrete_map={'Positive': '#2E8B57', 'Negative': '#DC143C'}
+        )
+        pie_html = fig_pie.to_html(include_plotlyjs='cdn', div_id="sentiment-pie")
+        
+        # Create confidence distribution chart
+        fig_hist = px.histogram(
+            df,
+            x='confidence',
+            nbins=20,
+            title="Confidence Score Distribution",
+            color='sentiment',
+            color_discrete_map={'Positive': '#2E8B57', 'Negative': '#DC143C'}
+        )
+        hist_html = fig_hist.to_html(include_plotlyjs='cdn', div_id="confidence-hist")
+        
+        # Generate recent predictions table
+        recent_df = df.head(10).copy()
+        recent_df['text'] = recent_df['text'].str[:100] + '...'
+        recent_df['confidence'] = recent_df['confidence'].apply(lambda x: f"{x:.2%}")
+        recent_df['created_at'] = recent_df['created_at'].dt.strftime('%Y-%m-%d %H:%M:%S')
+        
+        table_rows = ""
+        for _, row in recent_df.iterrows():
+            sentiment_class = "positive" if row['sentiment'] == 'Positive' else "negative"
+            table_rows += f"""
+            <tr>
+                <td>{row['created_at']}</td>
+                <td style="max-width: 300px;">{row['text']}</td>
+                <td><span class="sentiment {sentiment_class}">{row['sentiment']}</span></td>
+                <td>{row['confidence']}</td>
+            </tr>
+            """
+        
+        # HTML template
+        html_content = f"""
+        <!DOCTYPE html>
+        <html>
+        <head>
+            <title>NegaBot Analytics Dashboard</title>
+            <meta charset="utf-8">
+            <meta name="viewport" content="width=device-width, initial-scale=1">
+            <style>
+                body {{
+                    font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Roboto', sans-serif;
+                    margin: 0;
+                    padding: 20px;
+                    background-color: #f8f9fa;
+                }}
+                .container {{
+                    max-width: 1200px;
+                    margin: 0 auto;
+                }}
+                .header {{
+                    text-align: center;
+                    color: #1f77b4;
+                    margin-bottom: 30px;
+                }}
+                .metrics-grid {{
+                    display: grid;
+                    grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
+                    gap: 20px;
+                    margin-bottom: 30px;
+                }}
+                .metric-card {{
+                    background: white;
+                    padding: 20px;
+                    border-radius: 8px;
+                    box-shadow: 0 2px 4px rgba(0,0,0,0.1);
+                    text-align: center;
+                }}
+                .metric-value {{
+                    font-size: 2em;
+                    font-weight: bold;
+                    color: #1f77b4;
+                }}
+                .metric-label {{
+                    color: #666;
+                    margin-top: 5px;
+                }}
+                .charts-grid {{
+                    display: grid;
+                    grid-template-columns: 1fr 1fr;
+                    gap: 20px;
+                    margin-bottom: 30px;
+                }}
+                .chart-container {{
+                    background: white;
+                    padding: 20px;
+                    border-radius: 8px;
+                    box-shadow: 0 2px 4px rgba(0,0,0,0.1);
+                }}
+                .table-container {{
+                    background: white;
+                    padding: 20px;
+                    border-radius: 8px;
+                    box-shadow: 0 2px 4px rgba(0,0,0,0.1);
+                    overflow-x: auto;
+                }}
+                table {{
+                    width: 100%;
+                    border-collapse: collapse;
+                }}
+                th, td {{
+                    padding: 12px;
+                    text-align: left;
+                    border-bottom: 1px solid #eee;
+                }}
+                th {{
+                    background-color: #f8f9fa;
+                    font-weight: 600;
+                }}
+                .sentiment.positive {{
+                    background-color: #d4edda;
+                    color: #155724;
+                    padding: 4px 8px;
+                    border-radius: 4px;
+                    font-size: 0.9em;
+                }}
+                .sentiment.negative {{
+                    background-color: #f8d7da;
+                    color: #721c24;
+                    padding: 4px 8px;
+                    border-radius: 4px;
+                    font-size: 0.9em;
+                }}
+                .refresh-btn {{
+                    background-color: #1f77b4;
+                    color: white;
+                    border: none;
+                    padding: 10px 20px;
+                    border-radius: 4px;
+                    cursor: pointer;
+                    font-size: 14px;
+                    margin-bottom: 20px;
+                }}
+                .refresh-btn:hover {{
+                    background-color: #1865a0;
+                }}
+                .download-btn {{
+                    background-color: #28a745;
+                    color: white;
+                    text-decoration: none;
+                    padding: 8px 16px;
+                    border-radius: 4px;
+                    font-size: 14px;
+                    display: inline-block;
+                    transition: background-color 0.2s;
+                }}
+                .download-btn:hover {{
+                    background-color: #218838;
+                    text-decoration: none;
+                    color: white;
+                }}
+                @media (max-width: 768px) {{
+                    .charts-grid {{
+                        grid-template-columns: 1fr;
+                    }}
+                }}
+            </style>
+        </head>
+        <body>
+            <div class="container">
+                <div class="header">
+                    <h1>🤖 NegaBot Analytics Dashboard</h1>
+                    <button class="refresh-btn" onclick="location.reload()">🔄 Refresh Data</button>
+                </div>
+                
+                <div class="metrics-grid">
+                    <div class="metric-card">
+                        <div class="metric-value">{total_predictions}</div>
+                        <div class="metric-label">📊 Total Predictions</div>
+                    </div>
+                    <div class="metric-card">
+                        <div class="metric-value">{positive_count}</div>
+                        <div class="metric-label">😊 Positive</div>
+                    </div>
+                    <div class="metric-card">
+                        <div class="metric-value">{negative_count}</div>
+                        <div class="metric-label">😞 Negative</div>
+                    </div>
+                    <div class="metric-card">
+                        <div class="metric-value">{avg_confidence:.1%}</div>
+                        <div class="metric-label">🎯 Avg Confidence</div>
+                    </div>
+                </div>
+                
+                <div class="charts-grid">
+                    <div class="chart-container">
+                        {pie_html}
+                    </div>
+                    <div class="chart-container">
+                        {hist_html}
+                    </div>
+                </div>
+                
+                <div class="table-container">
+                    <h3>📝 Recent Predictions</h3>
+                    <div style="margin-bottom: 15px;">
+                        <a href="/download/predictions.csv" class="download-btn" style="margin-right: 10px;">📥 Download CSV</a>
+                        <a href="/download/predictions.json" class="download-btn">📥 Download JSON</a>
+                    </div>
+                    <table>
+                        <thead>
+                            <tr>
+                                <th>Timestamp</th>
+                                <th>Tweet Text</th>
+                                <th>Sentiment</th>
+                                <th>Confidence</th>
+                            </tr>
+                        </thead>
+                        <tbody>
+                            {table_rows}
+                        </tbody>
+                    </table>
+                </div>
+                
+                <div style="text-align: center; margin-top: 30px; color: #666; font-size: 0.9em;">
+                    🤖 NegaBot Analytics Dashboard | Last updated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
+                </div>
+            </div>
+        </body>
+        </html>
+        """
+        
+        return HTMLResponse(content=html_content)
+        
+    except Exception as e:
+        logger.error(f"Error generating dashboard: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Failed to generate dashboard: {str(e)}")
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)

database.py

@@ -0,0 +1,289 @@
+"""
+Database and Logging System for NegaBot API
+Handles prediction logging using SQLite database
+"""
+import sqlite3
+import json
+import logging
+from datetime import datetime
+from typing import List, Dict
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Database configuration
+DB_PATH = "negabot_predictions.db"
+
+class PredictionLogger:
+    def __init__(self, db_path: str = DB_PATH):
+        """
+        Initialize the prediction logger with SQLite database
+        
+        Args:
+            db_path (str): Path to SQLite database file
+        """
+        self.db_path = db_path
+        self.init_database()
+    
+    def init_database(self):
+        """Initialize the database with required tables"""
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                
+                # Create predictions table
+                cursor.execute("""
+                    CREATE TABLE IF NOT EXISTS predictions (
+                        id INTEGER PRIMARY KEY AUTOINCREMENT,
+                        text TEXT NOT NULL,
+                        sentiment TEXT NOT NULL,
+                        confidence REAL NOT NULL,
+                        predicted_class INTEGER NOT NULL,
+                        timestamp TEXT NOT NULL,
+                        metadata TEXT,
+                        created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+                    )
+                """)
+                
+                # Create index for faster queries
+                cursor.execute("""
+                    CREATE INDEX IF NOT EXISTS idx_sentiment ON predictions(sentiment)
+                """)
+                cursor.execute("""
+                    CREATE INDEX IF NOT EXISTS idx_timestamp ON predictions(timestamp)
+                """)
+                
+                conn.commit()
+                logger.info("Database initialized successfully")
+                
+        except Exception as e:
+            logger.error(f"Error initializing database: {str(e)}")
+            raise e
+    
+    def log_prediction(self, text: str, sentiment: str, confidence: float, 
+                      predicted_class: int = None, metadata: Dict = None):
+        """
+        Log a prediction to the database
+        
+        Args:
+            text (str): Input text
+            sentiment (str): Predicted sentiment
+            confidence (float): Prediction confidence
+            predicted_class (int): Predicted class (0 or 1)
+            metadata (dict): Optional metadata
+        """
+        try:
+            # Infer predicted_class if not provided
+            if predicted_class is None:
+                predicted_class = 1 if sentiment == "Negative" else 0
+            
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                
+                cursor.execute("""
+                    INSERT INTO predictions (text, sentiment, confidence, predicted_class, timestamp, metadata)
+                    VALUES (?, ?, ?, ?, ?, ?)
+                """, (
+                    text,
+                    sentiment,
+                    confidence,
+                    predicted_class,
+                    datetime.now().isoformat(),
+                    json.dumps(metadata) if metadata else None
+                ))
+                
+                conn.commit()
+                
+        except Exception as e:
+            logger.error(f"Error logging prediction: {str(e)}")
+            raise e
+    
+    def get_all_predictions(self, limit: int = None) -> List[Dict]:
+        """
+        Get all predictions from the database
+        
+        Args:
+            limit (int): Maximum number of records to return
+            
+        Returns:
+            List of prediction dictionaries
+        """
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                
+                query = """
+                    SELECT id, text, sentiment, confidence, predicted_class, timestamp, metadata, created_at
+                    FROM predictions 
+                    ORDER BY created_at DESC
+                """
+                
+                if limit:
+                    query += f" LIMIT {limit}"
+                
+                cursor.execute(query)
+                rows = cursor.fetchall()
+                
+                predictions = []
+                for row in rows:
+                    prediction = {
+                        "id": row[0],
+                        "text": row[1],
+                        "sentiment": row[2],
+                        "confidence": row[3],
+                        "predicted_class": row[4],
+                        "timestamp": row[5],
+                        "metadata": json.loads(row[6]) if row[6] else None,
+                        "created_at": row[7]
+                    }
+                    predictions.append(prediction)
+                
+                return predictions
+                
+        except Exception as e:
+            logger.error(f"Error getting predictions: {str(e)}")
+            return []
+    
+    def get_predictions_by_sentiment(self, sentiment: str) -> List[Dict]:
+        """
+        Get predictions filtered by sentiment
+        
+        Args:
+            sentiment (str): Sentiment to filter by ("Positive" or "Negative")
+            
+        Returns:
+            List of prediction dictionaries
+        """
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                
+                cursor.execute("""
+                    SELECT id, text, sentiment, confidence, predicted_class, timestamp, metadata, created_at
+                    FROM predictions 
+                    WHERE sentiment = ?
+                    ORDER BY created_at DESC
+                """, (sentiment,))
+                
+                rows = cursor.fetchall()
+                
+                predictions = []
+                for row in rows:
+                    prediction = {
+                        "id": row[0],
+                        "text": row[1],
+                        "sentiment": row[2],
+                        "confidence": row[3],
+                        "predicted_class": row[4],
+                        "timestamp": row[5],
+                        "metadata": json.loads(row[6]) if row[6] else None,
+                        "created_at": row[7]
+                    }
+                    predictions.append(prediction)
+                
+                return predictions
+                
+        except Exception as e:
+            logger.error(f"Error getting predictions by sentiment: {str(e)}")
+            return []
+    
+    def get_stats(self) -> Dict:
+        """
+        Get prediction statistics
+        
+        Returns:
+            Dictionary with statistics
+        """
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                
+                # Total count
+                cursor.execute("SELECT COUNT(*) FROM predictions")
+                total_count = cursor.fetchone()[0]
+                
+                if total_count == 0:
+                    return {
+                        "total_predictions": 0,
+                        "positive_count": 0,
+                        "negative_count": 0,
+                        "average_confidence": 0
+                    }
+                
+                # Sentiment counts
+                cursor.execute("SELECT sentiment, COUNT(*) FROM predictions GROUP BY sentiment")
+                sentiment_counts = dict(cursor.fetchall())
+                
+                # Average confidence
+                cursor.execute("SELECT AVG(confidence) FROM predictions")
+                avg_confidence = cursor.fetchone()[0]
+                
+                return {
+                    "total_predictions": total_count,
+                    "positive_count": sentiment_counts.get("Positive", 0),
+                    "negative_count": sentiment_counts.get("Negative", 0),
+                    "average_confidence": round(avg_confidence, 4) if avg_confidence else 0
+                }
+                
+        except Exception as e:
+            logger.error(f"Error getting stats: {str(e)}")
+            return {}
+
+# Global logger instance
+_logger_instance = None
+
+def get_logger():
+    """Get the global logger instance"""
+    global _logger_instance
+    if _logger_instance is None:
+        _logger_instance = PredictionLogger()
+    return _logger_instance
+
+def log_prediction(text: str, sentiment: str, confidence: float, metadata: Dict = None):
+    """Convenience function to log a prediction"""
+    logger_instance = get_logger()
+    logger_instance.log_prediction(text, sentiment, confidence, metadata=metadata)
+
+def get_all_predictions(limit: int = None) -> List[Dict]:
+    """Convenience function to get all predictions"""
+    logger_instance = get_logger()
+    return logger_instance.get_all_predictions(limit=limit)
+
+def get_predictions_by_sentiment(sentiment: str) -> List[Dict]:
+    """Convenience function to get predictions by sentiment"""
+    logger_instance = get_logger()
+    return logger_instance.get_predictions_by_sentiment(sentiment)
+
+def get_prediction_stats() -> Dict:
+    """Convenience function to get prediction statistics"""
+    logger_instance = get_logger()
+    return logger_instance.get_stats()
+
+if __name__ == "__main__":
+    # Test the logging system
+    logger_instance = PredictionLogger()
+    
+    # Test logging
+    test_predictions = [
+        ("This product is amazing!", "Positive", 0.95),
+        ("Terrible quality, waste of money", "Negative", 0.89),
+        ("It's okay, nothing special", "Positive", 0.67),
+        ("Awful customer service", "Negative", 0.92)
+    ]
+    
+    print("Testing prediction logging...")
+    for text, sentiment, confidence in test_predictions:
+        logger_instance.log_prediction(text, sentiment, confidence)
+        print(f"Logged: {sentiment} - {text}")
+    
+    # Test retrieval
+    print("\nRetrieving all predictions:")
+    predictions = logger_instance.get_all_predictions()
+    for pred in predictions:
+        print(f"ID: {pred['id']}, Sentiment: {pred['sentiment']}, Text: {pred['text'][:50]}...")
+    
+    # Test stats
+    print("\nPrediction statistics:")
+    stats = logger_instance.get_stats()
+    print(json.dumps(stats, indent=2))