app.py

import pandas as pd
import requests
from bs4 import BeautifulSoup
from transformers import pipeline
import logging
from datetime import datetime
import numpy as np
import gradio as gr

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Initialize sentiment analysis model
sentiment_analyzer = None

def load_sentiment_model():
    global sentiment_analyzer
    try:
        sentiment_analyzer = pipeline("sentiment-analysis", model="distilbert-base-uncased-finetuned-sst-2-english")
        logger.info("Sentiment analysis model successfully initialized.")
        return "Model loaded successfully!"
    except Exception as e:
        logger.error(f"Failed to initialize sentiment analysis model: {e}")
        return f"Error loading model: {str(e)}"

# Load the model at startup
load_sentiment_model()

def scrape_finviz_news(ticker):
    """
    Scrapes news headlines for the given ticker from Finviz.
    """
    try:
        url = f"https://finviz.com/quote.ashx?t={ticker}"
        headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}
        
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, 'html.parser')
        news_table = soup.find(id='news-table')
        
        if not news_table:
            return [], []
        
        headlines = []
        dates = []
        for row in news_table.findAll('tr'):
            headline = row.find('a', class_='tab-link-news')
            date_data = row.find('td', align='right')
            if headline and date_data:
                headlines.append(headline.text.strip())
                date_text = date_data.text.strip().split(' ')
                date = date_text[0] if len(date_text) > 0 else datetime.now().strftime('%Y-%m-%d')
                dates.append(date)
        
        logger.info(f"Scraped {len(headlines)} headlines for {ticker}.")
        return headlines, dates
    except Exception as e:
        logger.error(f"Error scraping {ticker}: {e}")
        return [], []
    
def analyze_sentiment(headlines):
    """
    Performs sentiment analysis on headlines.
    """
    try:
        if not headlines:
            return [], []
        
        results = sentiment_analyzer(headlines)
        labels = [res['label'] for res in results]
        scores = [res['score'] for res in results]
        return labels, scores
    except Exception as e:
        logger.error(f"Sentiment analysis error: {e}")
        return [], []

def analyze_stock(ticker):
    """
    Analyzes a single stock ticker and returns results.
    """
    try:
        ticker = ticker.strip().upper()
        if not ticker:
            return "Please enter a valid ticker symbol."
        
        headlines, dates = scrape_finviz_news(ticker)
        if not headlines:
            return f"No headlines found for {ticker}."
        
        labels, scores = analyze_sentiment(headlines)
        if not labels:
            return f"Could not analyze sentiment for {ticker}."
        
        positive_count = labels.count('POSITIVE')
        negative_count = labels.count('NEGATIVE')
        average_score = np.mean([score if label == 'POSITIVE' else -score for label, score in zip(labels, scores)])
        
        # Format results as text
        result = f"## Analysis for {ticker}\n\n"
        result += f"- Headlines analyzed: {len(headlines)}\n"
        result += f"- Average sentiment score: {average_score:.4f}\n"
        result += f"- Positive headlines: {positive_count}\n"
        result += f"- Negative headlines: {negative_count}\n\n"
        
        result += "### Recent Headlines:\n\n"
        for i, (headline, date, label, score) in enumerate(zip(headlines, dates, labels, scores), 1):
            if i > 10:  # Limit to 10 headlines
                break
            result += f"{i}. [{date}] {headline} - {label} ({score:.2f})\n"
        
        return result
    except Exception as e:
        logger.error(f"Error analyzing {ticker}: {e}")
        return f"Error analyzing {ticker}: {str(e)}"

# Create a simple Gradio interface
iface = gr.Interface(
    fn=analyze_stock,
    inputs=gr.Textbox(placeholder="Enter a ticker symbol (e.g., AAPL, MSFT, TSLA)"),
    outputs=gr.Markdown(),
    title="Stock Sentiment Analysis",
    description="Enter a stock ticker symbol to analyze news sentiment",
    examples=[["AAPL"], ["TSLA"], ["MSFT"], ["AMZN"], ["GOOGL"]],
    allow_flagging="never"
)

# Launch the app
if __name__ == "__main__":
    iface.launch()