app.py

import logging
import json
import gradio as gr
from transformers import pipeline
import torch

# Set up logging
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)

# Sentiment analysis model
SENTIMENT_ANALYSIS_MODEL = "mrm8488/distilroberta-finetuned-financial-news-sentiment-analysis"
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
logging.info(f"Using device: {DEVICE}")
logging.info("Initializing sentiment analysis model...")
sentiment_analyzer = pipeline(
    "sentiment-analysis", model=SENTIMENT_ANALYSIS_MODEL, device=DEVICE
)
logging.info("Model initialized successfully")

# Function to analyze sentiment of a single article
def analyze_article_sentiment(article: str) -> float:
    """
    Analyze sentiment for a single article and return a numerical score.
    Positive = 1, Neutral = 0, Negative = -1
    """
    sentiment = sentiment_analyzer(article)[0]
    label = sentiment["label"].lower()
    score = sentiment["score"]

    # Map sentiment to numerical values
    if label == "positive":
        return score
    elif label == "negative":
        return -score
    else:  # neutral
        return 0

# Function to calculate overall sentiment for a coin
def calculate_overall_sentiment(sentiment_scores: list) -> str:
    """
    Calculate the overall sentiment based on the average score:
    - Bullish: Average > 0.1
    - Bearish: Average < -0.1
    - Neutral: Otherwise
    """
    average_score = sum(sentiment_scores) / len(sentiment_scores)
    if average_score > 0.1:
        return "bullish"
    elif average_score < -0.1:
        return "bearish"
    else:
        return "neutral"

# Main function to process sentiment for multiple coins
def analyze_sentiment(input_json: str) -> dict:
    try:
        # Parse the input JSON
        data = json.loads(input_json)
        results = {}

        for coin_data in data["coins"]:
            coin_name = coin_data["coin"]
            articles = coin_data["articles"]
            logging.info(f"Analyzing sentiment for {coin_name} ({len(articles)} articles)")

            # Analyze sentiment for each article
            sentiment_scores = [analyze_article_sentiment(article["description"]) for article in articles]

            # Calculate overall sentiment
            overall_sentiment = calculate_overall_sentiment(sentiment_scores)
            results[coin_name] = overall_sentiment

            logging.info(f"{coin_name} sentiment: {overall_sentiment}")

        return {"results": results}
    except Exception as e:
        logging.error(f"Error during sentiment analysis: {e}")
        return {"error": "Failed to analyze sentiment"}

# Gradio Interface
with gr.Blocks() as iface:
    gr.Markdown("# Crypto Sentiment Analysis")
    gr.Markdown("Enter a JSON payload with news articles for multiple coins, and I'll analyze their sentiment!")

    with gr.Row():
        input_json = gr.Textbox(
            label="Input JSON",
            lines=10,
            placeholder="""{
  "coins": [
    {
      "coin": "BTC",
      "articles": [
        {"title": "Bitcoin Price Surges", "description": "Bitcoin's price surged above $30,000."},
        {"title": "Bitcoin Faces Challenges", "description": "Regulators are scrutinizing Bitcoin."}
      ]
    },
    {
      "coin": "XRP",
      "articles": [
        {"title": "XRP Gains Momentum", "description": "XRP's price rose after a favorable court ruling."},
        {"title": "XRP Faces Uncertainty", "description": "Traders remain cautious about XRP's future."}
      ]
    },
    {
      "coin": "ETH",
      "articles": [
        {"title": "Ethereum Upgrades Network", "description": "Ethereum completed its latest upgrade, improving scalability."},
        {"title": "Ethereum Faces Gas Fee Criticism", "description": "Users complain about high gas fees on Ethereum."}
      ]
    }
  ]
}"""
        )
    
    with gr.Row():
        analyze_button = gr.Button("Analyze Sentiment", size="sm")
    
    with gr.Row():
        output_json = gr.JSON(label="Sentiment Results")

    # Button click handler
    analyze_button.click(
        analyze_sentiment,
        inputs=[input_json],
        outputs=[output_json],
    )

# Launch the Gradio app
logging.info("Launching Gradio interface")
iface.queue().launch()