Spaces:
Sleeping
Sleeping
import streamlit as st | |
import requests | |
from bs4 import BeautifulSoup | |
from transformers import pipeline | |
from transformers import AutoTokenizer, AutoModelForSequenceClassification | |
import time | |
model_id = "LinkLinkWu/Boss_Stock_News_Analysis" | |
# Load tokenizer & Model | |
tokenizer = AutoTokenizer.from_pretrained(model_id) | |
model = AutoModelForSequenceClassification.from_pretrained(model_id) | |
# Initialize sentiment analysis pipeline | |
sentiment_pipeline = pipeline("sentiment-analysis", model=model, tokenizer=tokenizer) | |
# Function to fetch top 50 news articles from FinViz | |
def fetch_news(ticker): | |
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) | |
soup = BeautifulSoup(response.text, 'html.parser') | |
news_table = soup.find(id='news-table') | |
news = [] | |
for row in news_table.findAll('tr')[:50]: # Fetch up to 50 articles | |
title = row.a.get_text() | |
link = row.a['href'] | |
news.append({'title': title, 'link': link}) | |
return news | |
except Exception as e: | |
st.error(f"Failed to fetch news for {ticker}: {e}") | |
return [] | |
# Function to analyze sentiment of news title | |
def analyze_sentiment(text): | |
try: | |
result = sentiment_pipeline(text)[0] | |
return "Positive" if result['label'] == 'POSITIVE' else "Negative" | |
except Exception as e: | |
st.error(f"Sentiment analysis failed: {e}") | |
return "Unknown" | |
# Streamlit UI | |
st.title("Stock News Sentiment Analysis") | |
# Input field for stock tickers | |
tickers_input = st.text_input("Enter five stock tickers separated by commas (e.g., AAPL, MSFT, GOOGL, AMZN, TSLA):") | |
if st.button("Get News and Sentiment"): | |
if tickers_input: | |
tickers = [ticker.strip().upper() for ticker in tickers_input.split(',')] | |
# Validate input | |
if len(tickers) != 5: | |
st.error("Please enter exactly five stock tickers.") | |
else: | |
progress_bar = st.progress(0) | |
total_stocks = len(tickers) | |
for idx, ticker in enumerate(tickers): | |
st.subheader(f"Analyzing {ticker}...") | |
news_list = fetch_news(ticker) | |
if news_list: | |
# Analyze sentiment for all news articles (up to 50) | |
sentiments = [] | |
for news in news_list: | |
sentiment = analyze_sentiment(news['title']) | |
sentiments.append(sentiment) | |
# Determine overall sentiment based on majority | |
positive_count = sentiments.count("Positive") | |
negative_count = sentiments.count("Negative") | |
overall_sentiment = "Positive" if positive_count > negative_count else "Negative" | |
# Display top 3 news articles with sentiment | |
st.write(f"**Top 3 News Articles for {ticker}**") | |
for i, news in enumerate(news_list[:3], 1): | |
sentiment = sentiments[i-1] | |
st.markdown(f"{i}. [{news['title']}]({news['link']}) - **{sentiment}**") | |
# Display overall sentiment | |
st.write(f"**Overall Sentiment for {ticker}: {overall_sentiment}**") | |
else: | |
st.write(f"No news available for {ticker}.") | |
# Update progress bar | |
progress_bar.progress((idx + 1) / total_stocks) | |
time.sleep(0.1) # Simulate processing time | |
else: | |
st.warning("Please enter stock tickers.") |