import httpx from typing import List, Dict from datetime import datetime, timedelta from textblob import TextBlob import hashlib from app.config import settings from app.schemas.schemas import ( NewsArticle, NewsFeedResponse, Sentiment, EconomicEvent, EconomicCalendarResponse, ) class NewsService: def __init__(self): self.alpha_vantage_key = settings.ALPHA_VANTAGE_API_KEY self.finnhub_key = settings.FINNHUB_API_KEY self.news_api_key = settings.NEWS_API_KEY # Gold-related keywords for relevance scoring self.gold_keywords = { "high_relevance": [ "gold", "xau", "precious metals", "bullion", "gold price", "gold market", "gold trading", "gold miners", "gold etf" ], "medium_relevance": [ "federal reserve", "fed", "inflation", "interest rates", "dollar", "usd", "monetary policy", "central bank", "jerome powell", "treasury", "bonds" ], "context_relevance": [ "geopolitics", "war", "sanctions", "recession", "crisis", "safe haven", "risk off", "uncertainty" ] } # Impact categories self.impact_categories = { "MONETARY_POLICY": ["federal reserve", "fed", "interest rate", "monetary policy", "central bank"], "GEOPOLITICS": ["war", "conflict", "sanctions", "tension", "geopolitical"], "ECONOMIC_DATA": ["inflation", "cpi", "gdp", "employment", "jobs", "unemployment"], "MARKET_SENTIMENT": ["risk", "sentiment", "volatility", "safe haven"], "COMMODITY": ["gold", "precious metals", "bullion", "commodities"], } def _calculate_relevance_score(self, text: str) -> float: """Calculate how relevant a news article is to gold trading""" text_lower = text.lower() score = 0.0 # High relevance keywords for keyword in self.gold_keywords["high_relevance"]: if keyword in text_lower: score += 0.4 # Medium relevance keywords for keyword in self.gold_keywords["medium_relevance"]: if keyword in text_lower: score += 0.2 # Context relevance keywords for keyword in self.gold_keywords["context_relevance"]: if keyword in text_lower: score += 0.1 return min(score, 1.0) def _categorize_news(self, text: str) -> str: """Categorize news based on content""" text_lower = text.lower() for category, keywords in self.impact_categories.items(): for keyword in keywords: if keyword in text_lower: return category return "OTHER" def _analyze_sentiment(self, text: str) -> tuple[Sentiment, float]: """Analyze sentiment using TextBlob""" try: analysis = TextBlob(text) polarity = analysis.sentiment.polarity if polarity > 0.1: sentiment = Sentiment.POSITIVE elif polarity < -0.1: sentiment = Sentiment.NEGATIVE else: sentiment = Sentiment.NEUTRAL return sentiment, polarity except Exception: return Sentiment.NEUTRAL, 0.0 def _assess_gold_impact(self, sentiment: Sentiment, category: str, relevance: float) -> str: """Assess impact level on gold prices""" # High impact categories high_impact_cats = ["MONETARY_POLICY", "ECONOMIC_DATA"] if relevance > 0.7: if category in high_impact_cats: return "HIGH" return "MEDIUM" elif relevance > 0.4: return "MEDIUM" else: return "LOW" async def fetch_alpha_vantage_news(self, topics: str = "economy_monetary,finance") -> List[NewsArticle]: """Fetch news from Alpha Vantage News Sentiment API""" try: params = { "function": "NEWS_SENTIMENT", "topics": topics, "limit": 50, "apikey": self.alpha_vantage_key, } async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get( settings.ALPHA_VANTAGE_BASE_URL, params=params ) response.raise_for_status() data = response.json() if "feed" not in data: return [] articles = [] for item in data["feed"]: title = item.get("title", "") summary = item.get("summary", "") full_text = f"{title} {summary}" relevance = self._calculate_relevance_score(full_text) # Filter only gold-relevant news if relevance < 0.3: continue sentiment, score = self._analyze_sentiment(full_text) category = self._categorize_news(full_text) impact = self._assess_gold_impact(sentiment, category, relevance) # Parse published date published_str = item.get("time_published", "") try: published_at = datetime.strptime(published_str, "%Y%m%dT%H%M%S") except: published_at = datetime.now() article_id = hashlib.md5(f"{title}{published_str}".encode()).hexdigest() articles.append( NewsArticle( id=article_id, source=item.get("source", "Alpha Vantage"), title=title, description=summary, url=item.get("url", ""), published_at=published_at, sentiment=sentiment, sentiment_score=score, impact_on_gold=impact, relevance_score=relevance, category=category, ) ) return articles except Exception as e: print(f"Error fetching Alpha Vantage news: {e}") return [] async def fetch_finnhub_news(self) -> List[NewsArticle]: """Fetch gold-related news from Finnhub""" if not self.finnhub_key: return [] try: # Get general market news params = { "category": "forex", "token": self.finnhub_key, } async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get( f"{settings.FINNHUB_BASE_URL}/news", params=params ) response.raise_for_status() data = response.json() articles = [] for item in data[:50]: # Limit to 50 articles title = item.get("headline", "") summary = item.get("summary", "") full_text = f"{title} {summary}" relevance = self._calculate_relevance_score(full_text) # Filter only gold-relevant news if relevance < 0.3: continue sentiment, score = self._analyze_sentiment(full_text) category = self._categorize_news(full_text) impact = self._assess_gold_impact(sentiment, category, relevance) published_at = datetime.fromtimestamp(item.get("datetime", 0)) article_id = hashlib.md5(f"{title}{item.get('id', '')}".encode()).hexdigest() articles.append( NewsArticle( id=article_id, source=item.get("source", "Finnhub"), title=title, description=summary, url=item.get("url", ""), published_at=published_at, sentiment=sentiment, sentiment_score=score, impact_on_gold=impact, relevance_score=relevance, category=category, ) ) return articles except Exception as e: print(f"Error fetching Finnhub news: {e}") return [] async def get_aggregated_news_feed(self) -> NewsFeedResponse: """Get aggregated news from all sources""" # Fetch from multiple sources alpha_news = await self.fetch_alpha_vantage_news() finnhub_news = await self.fetch_finnhub_news() if self.finnhub_key else [] # Combine and deduplicate all_articles = alpha_news + finnhub_news # Remove duplicates based on similar titles unique_articles = [] seen_titles = set() for article in all_articles: title_key = article.title.lower()[:50] # First 50 chars if title_key not in seen_titles: seen_titles.add(title_key) unique_articles.append(article) # Sort by published date (newest first) unique_articles.sort(key=lambda x: x.published_at, reverse=True) # Limit to most recent 50 unique_articles = unique_articles[:50] # Calculate statistics bullish_count = sum(1 for a in unique_articles if a.sentiment == Sentiment.POSITIVE) bearish_count = sum(1 for a in unique_articles if a.sentiment == Sentiment.NEGATIVE) neutral_count = sum(1 for a in unique_articles if a.sentiment == Sentiment.NEUTRAL) avg_sentiment = ( sum(a.sentiment_score for a in unique_articles) / len(unique_articles) if unique_articles else 0.0 ) # Determine overall sentiment if avg_sentiment > 0.1: overall_sentiment = Sentiment.POSITIVE elif avg_sentiment < -0.1: overall_sentiment = Sentiment.NEGATIVE else: overall_sentiment = Sentiment.NEUTRAL return NewsFeedResponse( articles=unique_articles, total_count=len(unique_articles), bullish_count=bullish_count, bearish_count=bearish_count, neutral_count=neutral_count, overall_sentiment=overall_sentiment, avg_sentiment_score=avg_sentiment, ) async def get_economic_calendar(self) -> EconomicCalendarResponse: """Get upcoming economic events that impact gold""" # This would integrate with economic calendar APIs # For MVP, return curated list of upcoming events # In production, integrate with: # - Forex Factory API # - Investing.com Economic Calendar # - Alpha Vantage Economic Indicators # For now, return empty with structure events = [] # Count high-impact upcoming events now = datetime.now() upcoming_high_impact = sum( 1 for e in events if e.importance == "HIGH" and e.event_date > now ) return EconomicCalendarResponse( events=events, upcoming_high_impact=upcoming_high_impact, ) news_service = NewsService()