Spaces:
Running
Running
File size: 3,914 Bytes
dc3659d 23323a5 dc3659d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
import logging
import gradio as gr
import pandas as pd
import torch
from GoogleNews import GoogleNews
from transformers import pipeline
# Set up logging
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
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")
def fetch_articles(query):
try:
logging.info(f"Fetching articles for query: '{query}'")
googlenews = GoogleNews(lang="en")
googlenews.search(query)
articles = googlenews.result()
logging.info(f"Fetched {len(articles)} articles")
return articles
except Exception as e:
logging.error(
f"Error while searching articles for query: '{query}'. Error: {e}"
)
raise gr.Error(
f"Unable to search articles for query: '{query}'. Try again later...",
duration=5,
)
def analyze_article_sentiment(article):
logging.info(f"Analyzing sentiment for article: {article['title']}")
sentiment = sentiment_analyzer(article["desc"])[0]
article["sentiment"] = sentiment
return article
def analyze_asset_sentiment(asset_name):
logging.info(f"Starting sentiment analysis for asset: {asset_name}")
logging.info("Fetching articles")
articles = fetch_articles(asset_name)
logging.info("Analyzing sentiment of each article")
analyzed_articles = [analyze_article_sentiment(article) for article in articles]
logging.info("Sentiment analysis completed")
return convert_to_dataframe(analyzed_articles)
def convert_to_dataframe(analyzed_articles):
df = pd.DataFrame(analyzed_articles)
df["Title"] = df.apply(
lambda row: f'<a href="{row["link"]}" target="_blank">{row["title"]}</a>',
axis=1,
)
df["Description"] = df["desc"]
df["Date"] = df["date"]
def sentiment_badge(sentiment):
colors = {
"negative": "red",
"neutral": "gray",
"positive": "green",
}
color = colors.get(sentiment, "grey")
return f'<span style="background-color: {color}; color: white; padding: 2px 6px; border-radius: 4px;">{sentiment}</span>'
df["Sentiment"] = df["sentiment"].apply(lambda x: sentiment_badge(x["label"]))
return df[["Sentiment", "Title", "Description", "Date"]]
with gr.Blocks() as iface:
gr.Markdown("# Trading Asset Sentiment Analysis")
gr.Markdown(
"Enter the name of a trading asset, and I'll fetch recent articles and analyze their sentiment!"
)
with gr.Row():
input_asset = gr.Textbox(
label="Asset Name",
lines=1,
placeholder="Enter the name of the trading asset...",
)
with gr.Row():
analyze_button = gr.Button("Analyze Sentiment", size="sm")
gr.Examples(
examples=[
"Bitcoin",
"Tesla",
"Apple",
"Amazon",
],
inputs=input_asset,
)
with gr.Row():
with gr.Column():
with gr.Blocks():
gr.Markdown("## Articles and Sentiment Analysis")
articles_output = gr.Dataframe(
headers=["Sentiment", "Title", "Description", "Date"],
datatype=["markdown", "html", "markdown", "markdown"],
wrap=False,
)
analyze_button.click(
analyze_asset_sentiment,
inputs=[input_asset],
outputs=[articles_output],
)
logging.info("Launching Gradio interface")
iface.queue().launch()
|