Spaces:
Sleeping
Sleeping
| # app_news_sentiment_firebase.py | |
| import feedparser | |
| import pandas as pd | |
| import gradio as gr | |
| from transformers import pipeline | |
| from datetime import datetime | |
| import firebase_admin | |
| from firebase_admin import credentials, db | |
| import os | |
| import json | |
| # --- Firebase Configuration --- | |
| # IMPORTANT: Set these environment variables in your deployment environment (e.g., Hugging Face Spaces Secrets) | |
| # 1. FIRESTORE_SA_KEY: The full JSON content of your Firebase service account key. | |
| # 2. FIREBASE_DB_URL: The URL to your Firebase Realtime Database. | |
| SA_KEY_JSON = os.environ.get('FIRESTORE_SA_KEY') | |
| DB_URL = os.environ.get('FIREBASE_DB_URL') | |
| class NewsRTDBLogger: | |
| def __init__(self, db_ref_name='news_sentiment'): | |
| self.ref = None | |
| if not all([SA_KEY_JSON, DB_URL]): | |
| print("NEWS LOGGER: Firebase secrets not set. Logger is disabled.") | |
| return | |
| try: | |
| # The service account key is loaded from a JSON string in the environment variable | |
| cred_dict = json.loads(SA_KEY_JSON) | |
| cred = credentials.Certificate(cred_dict) | |
| if not firebase_admin._apps: | |
| firebase_admin.initialize_app(cred, {'databaseURL': DB_URL}) | |
| self.ref = db.reference(db_ref_name) | |
| print(f"NEWS LOGGER: Successfully connected to Firebase RTDB at '{db_ref_name}'.") | |
| except Exception as e: | |
| print(f"NEWS LOGGER: CRITICAL ERROR - Failed to initialize: {e}") | |
| def log_sentiment(self, pub_date, headline, sentiment, confidence): | |
| if not self.ref: | |
| return | |
| try: | |
| # Create a unique key based on the timestamp to keep logs ordered | |
| log_entry = { | |
| "timestamp_published": pub_date, | |
| "headline": headline, | |
| "sentiment": sentiment, | |
| "confidence_score": float(confidence) | |
| } | |
| self.ref.push(log_entry) | |
| # print(f"NEWS LOGGER: Logged '{headline}'") | |
| except Exception as e: | |
| print(f"NEWS LOGGER: CRITICAL ERROR - Could not write sentiment log: {e}") | |
| # --- Initialize Global Objects --- | |
| sentiment_pipeline = pipeline("text-classification", model="cardiffnlp/twitter-roberta-base-sentiment-latest") | |
| news_logger = NewsRTDBLogger() # Initialize the logger | |
| # --- Core Functions --- | |
| def fetch_news(): | |
| """Fetches the latest news headlines from the FXStreet RSS feed.""" | |
| url = "https://www.fxstreet.com/rss/news" | |
| feed = feedparser.parse(url) | |
| # Fetch a few more headlines to ensure we get recent, unique ones | |
| headlines = feed.entries[:15] | |
| return headlines | |
| def analyze_news(): | |
| """ | |
| Analyzes news sentiment, logs it to Firebase, and returns a DataFrame for display. | |
| """ | |
| headlines = fetch_news() | |
| results = [] | |
| for h in headlines: | |
| text = h.title | |
| # Get the publication date, fallback to now if not present | |
| pub_date = h.get("published", datetime.utcnow().isoformat()) | |
| # Perform sentiment analysis | |
| pred = sentiment_pipeline(text)[0] | |
| label = pred["label"].lower() | |
| score = round(pred["score"], 3) | |
| if label == "positive": | |
| market_sentiment = "Bullish" | |
| elif label == "negative": | |
| market_sentiment = "Bearish" | |
| else: | |
| market_sentiment = "Neutral" | |
| # Log the result to Firebase | |
| news_logger.log_sentiment(pub_date, text, market_sentiment, score) | |
| results.append([pub_date, text, market_sentiment, score]) | |
| # Create DataFrame for Gradio output, showing the 7 most recent | |
| df = pd.DataFrame(results, columns=["Date", "Headline", "Market Sentiment", "Confidence"]) | |
| return df.head(7) | |
| # --- Gradio Interface --- | |
| demo = gr.Interface( | |
| fn=analyze_news, | |
| inputs=None, | |
| outputs=gr.Dataframe(headers=["Date", "Headline", "Market Sentiment", "Confidence"]), | |
| title="📊 Forex Sentiment Agent (Firebase Edition)", | |
| description="Live Forex headlines sentiment (Bullish / Bearish / Neutral). Now logging results directly to Firebase.", | |
| allow_flagging='never' | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch() |