Spaces:
Sleeping
Sleeping
| import logging | |
| import os | |
| import tempfile | |
| from datetime import datetime | |
| import gradio as gr | |
| import pandas as pd | |
| import plotly.express as px | |
| import torch | |
| import yfinance as yf | |
| from GoogleNews import GoogleNews | |
| from transformers import pipeline | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s - %(levelname)s - %(message)s" | |
| ) | |
| SENTIMENT_ANALYSIS_MODEL = ( | |
| "mrm8488/distilroberta-finetuned-financial-news-sentiment-analysis" | |
| ) | |
| DEVICE = 0 if torch.cuda.is_available() else -1 | |
| logging.info(f"Using device: {'cuda' if DEVICE == 0 else 'cpu'}") | |
| sentiment_analyzer = pipeline( | |
| "sentiment-analysis", | |
| model=SENTIMENT_ANALYSIS_MODEL, | |
| device=DEVICE | |
| ) | |
| def fetch_articles(query, max_articles=20): | |
| try: | |
| googlenews = GoogleNews(lang="en") | |
| googlenews.search(query) | |
| articles = googlenews.result() | |
| if not articles: | |
| return [] | |
| df = pd.DataFrame(articles) | |
| required_cols = ["title", "desc", "link", "date", "media"] | |
| for col in required_cols: | |
| if col not in df.columns: | |
| df[col] = "" | |
| df = df.drop_duplicates(subset=["title", "link"]) | |
| df = df.head(max_articles) | |
| return df.to_dict("records") | |
| except Exception as e: | |
| logging.error(f"News fetch failed for {query}: {e}") | |
| raise gr.Error("Unable to fetch news. Try again later.") | |
| def analyze_sentiments_batch(articles): | |
| if not articles: | |
| return [] | |
| texts = [ | |
| f"{article.get('title', '')}. {article.get('desc', '')}" | |
| for article in articles | |
| ] | |
| sentiments = sentiment_analyzer( | |
| texts, | |
| batch_size=8, | |
| truncation=True, | |
| max_length=512 | |
| ) | |
| for article, sentiment in zip(articles, sentiments): | |
| article["sentiment_label"] = sentiment["label"].lower() | |
| article["sentiment_score"] = round(float(sentiment["score"]), 4) | |
| return articles | |
| def get_price_context(asset): | |
| try: | |
| ticker = yf.Ticker(asset) | |
| hist = ticker.history(period="1mo") | |
| if hist.empty: | |
| return "No price data found. Try using ticker symbols like AAPL, TSLA, BTC-USD, ETH-USD." | |
| latest_price = hist["Close"].iloc[-1] | |
| prev_price = hist["Close"].iloc[-2] if len(hist) > 1 else latest_price | |
| first_price = hist["Close"].iloc[0] | |
| one_day_return = ((latest_price - prev_price) / prev_price) * 100 | |
| one_month_return = ((latest_price - first_price) / first_price) * 100 | |
| volatility = hist["Close"].pct_change().std() * 100 | |
| return ( | |
| f"### Price Context\n" | |
| f"- Latest close: **{latest_price:.2f}**\n" | |
| f"- 1D return: **{one_day_return:.2f}%**\n" | |
| f"- 1M return: **{one_month_return:.2f}%**\n" | |
| f"- 1M daily volatility: **{volatility:.2f}%**" | |
| ) | |
| except Exception as e: | |
| logging.error(f"Price fetch failed for {asset}: {e}") | |
| return "Price data unavailable for this asset." | |
| def sentiment_badge(label): | |
| colors = { | |
| "negative": "#dc2626", | |
| "neutral": "#6b7280", | |
| "positive": "#16a34a", | |
| } | |
| color = colors.get(label.lower(), "#6b7280") | |
| return ( | |
| f'<span style="background-color:{color}; color:white; ' | |
| f'padding:4px 8px; border-radius:6px; font-weight:600;">' | |
| f'{label.upper()}</span>' | |
| ) | |
| def build_article_table(articles): | |
| if not articles: | |
| return pd.DataFrame() | |
| df = pd.DataFrame(articles) | |
| df["Sentiment"] = df["sentiment_label"].apply(sentiment_badge) | |
| df["Confidence"] = df["sentiment_score"] | |
| df["Title"] = df.apply( | |
| lambda row: f'<a href="{row["link"]}" target="_blank">{row["title"]}</a>', | |
| axis=1 | |
| ) | |
| df["Description"] = df["desc"] | |
| df["Source"] = df["media"] | |
| df["Date"] = df["date"] | |
| return df[ | |
| [ | |
| "Sentiment", | |
| "Confidence", | |
| "Title", | |
| "Description", | |
| "Source", | |
| "Date", | |
| ] | |
| ] | |
| def create_sentiment_summary(articles): | |
| if not articles: | |
| return "No articles found." | |
| df = pd.DataFrame(articles) | |
| counts = df["sentiment_label"].value_counts(normalize=True) * 100 | |
| avg_confidence = df["sentiment_score"].mean() | |
| positive = counts.get("positive", 0) | |
| negative = counts.get("negative", 0) | |
| neutral = counts.get("neutral", 0) | |
| bullish_score = positive - negative | |
| if bullish_score > 20: | |
| verdict = "Bullish" | |
| elif bullish_score < -20: | |
| verdict = "Bearish" | |
| else: | |
| verdict = "Neutral / Mixed" | |
| return ( | |
| f"### Sentiment Summary\n" | |
| f"- Overall view: **{verdict}**\n" | |
| f"- Positive: **{positive:.1f}%**\n" | |
| f"- Neutral: **{neutral:.1f}%**\n" | |
| f"- Negative: **{negative:.1f}%**\n" | |
| f"- Average model confidence: **{avg_confidence:.2f}**\n" | |
| f"- Bullish score: **{bullish_score:.1f}**" | |
| ) | |
| def generate_llm_style_summary(asset, articles): | |
| if not articles: | |
| return "No summary available." | |
| df = pd.DataFrame(articles) | |
| top_sentiment = df["sentiment_label"].value_counts().idxmax() | |
| top_articles = df.head(5) | |
| headlines = "\n".join( | |
| [f"- {row['title']}" for _, row in top_articles.iterrows()] | |
| ) | |
| return ( | |
| f"### AI-Style Market Brief\n" | |
| f"The current news sentiment for **{asset}** appears mostly " | |
| f"**{top_sentiment.upper()}** based on recent headlines.\n\n" | |
| f"Key headlines influencing this reading:\n" | |
| f"{headlines}\n\n" | |
| f"This should not be treated as a trading signal by itself. " | |
| f"Use it alongside price action, volume, volatility, macro events, " | |
| f"and risk management." | |
| ) | |
| def create_sentiment_chart(articles): | |
| if not articles: | |
| return None | |
| df = pd.DataFrame(articles) | |
| sentiment_counts = df["sentiment_label"].value_counts().reset_index() | |
| sentiment_counts.columns = ["Sentiment", "Count"] | |
| fig = px.bar( | |
| sentiment_counts, | |
| x="Sentiment", | |
| y="Count", | |
| title="Sentiment Distribution" | |
| ) | |
| return fig | |
| def create_source_chart(articles): | |
| if not articles: | |
| return None | |
| df = pd.DataFrame(articles) | |
| if "media" not in df.columns: | |
| return None | |
| source_counts = df["media"].replace("", "Unknown").value_counts().head(10) | |
| source_df = source_counts.reset_index() | |
| source_df.columns = ["Source", "Count"] | |
| fig = px.bar( | |
| source_df, | |
| x="Source", | |
| y="Count", | |
| title="Top News Sources" | |
| ) | |
| return fig | |
| def export_csv(articles): | |
| if not articles: | |
| return None | |
| df = pd.DataFrame(articles) | |
| filename = f"sentiment_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" | |
| filepath = os.path.join(tempfile.gettempdir(), filename) | |
| df.to_csv(filepath, index=False) | |
| return filepath | |
| def analyze_single_asset(asset_name, max_articles): | |
| if not asset_name.strip(): | |
| raise gr.Error("Please enter an asset name or ticker.") | |
| articles = fetch_articles(asset_name, max_articles) | |
| analyzed_articles = analyze_sentiments_batch(articles) | |
| article_table = build_article_table(analyzed_articles) | |
| sentiment_summary = create_sentiment_summary(analyzed_articles) | |
| price_context = get_price_context(asset_name) | |
| ai_summary = generate_llm_style_summary(asset_name, analyzed_articles) | |
| sentiment_chart = create_sentiment_chart(analyzed_articles) | |
| source_chart = create_source_chart(analyzed_articles) | |
| csv_file = export_csv(analyzed_articles) | |
| return ( | |
| sentiment_summary, | |
| price_context, | |
| ai_summary, | |
| article_table, | |
| sentiment_chart, | |
| source_chart, | |
| csv_file, | |
| ) | |
| def analyze_multiple_assets(asset_list, max_articles): | |
| assets = [ | |
| asset.strip() | |
| for asset in asset_list.split(",") | |
| if asset.strip() | |
| ] | |
| if not assets: | |
| raise gr.Error("Please enter at least one asset.") | |
| rows = [] | |
| for asset in assets: | |
| articles = fetch_articles(asset, max_articles) | |
| analyzed_articles = analyze_sentiments_batch(articles) | |
| if not analyzed_articles: | |
| rows.append( | |
| { | |
| "Asset": asset, | |
| "Overall Sentiment": "No data", | |
| "Positive %": 0, | |
| "Neutral %": 0, | |
| "Negative %": 0, | |
| "Bullish Score": 0, | |
| "Articles": 0, | |
| } | |
| ) | |
| continue | |
| df = pd.DataFrame(analyzed_articles) | |
| counts = df["sentiment_label"].value_counts(normalize=True) * 100 | |
| positive = counts.get("positive", 0) | |
| neutral = counts.get("neutral", 0) | |
| negative = counts.get("negative", 0) | |
| bullish_score = positive - negative | |
| if bullish_score > 20: | |
| overall = "Bullish" | |
| elif bullish_score < -20: | |
| overall = "Bearish" | |
| else: | |
| overall = "Neutral / Mixed" | |
| rows.append( | |
| { | |
| "Asset": asset, | |
| "Overall Sentiment": overall, | |
| "Positive %": round(positive, 2), | |
| "Neutral %": round(neutral, 2), | |
| "Negative %": round(negative, 2), | |
| "Bullish Score": round(bullish_score, 2), | |
| "Articles": len(analyzed_articles), | |
| } | |
| ) | |
| result_df = pd.DataFrame(rows) | |
| fig = px.bar( | |
| result_df, | |
| x="Asset", | |
| y="Bullish Score", | |
| title="Bullish Score Comparison" | |
| ) | |
| return result_df, fig | |
| with gr.Blocks(title="FinSentinel") as app: | |
| gr.Markdown("# FinSentinel") | |
| gr.Markdown( | |
| "Financial news sentiment dashboard for stocks, crypto, ETFs, commodities, and market themes." | |
| ) | |
| with gr.Tabs(): | |
| with gr.Tab("Single Asset Analysis"): | |
| with gr.Row(): | |
| asset_input = gr.Textbox( | |
| label="Asset Name or Ticker", | |
| placeholder="Example: AAPL, TSLA, Bitcoin, BTC-USD, Nvidia", | |
| ) | |
| max_articles = gr.Slider( | |
| minimum=5, | |
| maximum=50, | |
| value=20, | |
| step=5, | |
| label="Number of Articles", | |
| ) | |
| analyze_button = gr.Button("Analyze Asset", variant="primary") | |
| gr.Examples( | |
| examples=[ | |
| "AAPL", | |
| "TSLA", | |
| "NVDA", | |
| "BTC-USD", | |
| "Bitcoin", | |
| "Gold", | |
| ], | |
| inputs=asset_input, | |
| ) | |
| with gr.Row(): | |
| sentiment_summary_output = gr.Markdown() | |
| price_context_output = gr.Markdown() | |
| ai_summary_output = gr.Markdown() | |
| with gr.Row(): | |
| sentiment_chart_output = gr.Plot() | |
| source_chart_output = gr.Plot() | |
| articles_output = gr.Dataframe( | |
| label="Articles and Sentiment Analysis", | |
| headers=[ | |
| "Sentiment", | |
| "Confidence", | |
| "Title", | |
| "Description", | |
| "Source", | |
| "Date", | |
| ], | |
| datatype=[ | |
| "html", | |
| "number", | |
| "html", | |
| "str", | |
| "str", | |
| "str", | |
| ], | |
| wrap=True, | |
| interactive=False, | |
| ) | |
| csv_output = gr.File(label="Download CSV") | |
| analyze_button.click( | |
| analyze_single_asset, | |
| inputs=[asset_input, max_articles], | |
| outputs=[ | |
| sentiment_summary_output, | |
| price_context_output, | |
| ai_summary_output, | |
| articles_output, | |
| sentiment_chart_output, | |
| source_chart_output, | |
| csv_output, | |
| ], | |
| ) | |
| with gr.Tab("Portfolio Comparison"): | |
| portfolio_input = gr.Textbox( | |
| label="Assets", | |
| placeholder="Example: AAPL, TSLA, NVDA, BTC-USD, ETH-USD", | |
| lines=2, | |
| ) | |
| portfolio_articles = gr.Slider( | |
| minimum=5, | |
| maximum=30, | |
| value=10, | |
| step=5, | |
| label="Articles Per Asset", | |
| ) | |
| portfolio_button = gr.Button("Compare Assets", variant="primary") | |
| portfolio_table = gr.Dataframe( | |
| label="Portfolio Sentiment Comparison", | |
| interactive=False, | |
| ) | |
| portfolio_chart = gr.Plot() | |
| portfolio_button.click( | |
| analyze_multiple_assets, | |
| inputs=[portfolio_input, portfolio_articles], | |
| outputs=[portfolio_table, portfolio_chart], | |
| ) | |
| if __name__ == "__main__": | |
| app.queue().launch(server_name="0.0.0.0", server_port=7860) | |