""" Sentiment Analysis Engine. Provides sentiment analysis from multiple sources: - News headlines (via RSS feeds from Yahoo Finance, Google News) - Rule-based keyword scoring for fast fallback - LLM-based sentiment classification via Groq for nuanced analysis Uses financial keyword matching as primary method with optional LLM upgrade for more accurate sentiment classification. """ from __future__ import annotations import logging import re from datetime import datetime from typing import Any, Dict, List, Optional import aiohttp from xml.etree import ElementTree logger = logging.getLogger(__name__) from app.redis_client import cache_get, cache_set SENTIMENT_CACHE_TTL = 600 # 10 minutes # ── News RSS Sources ───────────────────────────────────────────────────── NEWS_FEEDS = { "yahoo": "https://feeds.finance.yahoo.com/rss/2.0/headline?s={ticker}®ion=US&lang=en-US", "google": "https://news.google.com/rss/search?q={ticker}+stock&hl=en-US&gl=US&ceid=US:en", } # Positive/negative financial keywords for rule-based sentiment POSITIVE_WORDS = { "surge", "soar", "jump", "gain", "rally", "rise", "bull", "record", "high", "profit", "beat", "outperform", "upgrade", "buy", "strong", "growth", "boom", "breakout", "momentum", "upside", "recovery", "optimistic", "bullish", } NEGATIVE_WORDS = { "crash", "plunge", "drop", "fall", "decline", "bear", "loss", "miss", "downgrade", "sell", "weak", "recession", "fear", "risk", "warning", "correction", "slump", "tumble", "underperform", "bearish", "concern", "layoff", "debt", "default", "bankruptcy", "investigation", "fraud", } def _score_headline(text: str) -> float: """Score a headline -1 to +1 using financial keyword matching.""" words = set(re.findall(r'\w+', text.lower())) pos = len(words & POSITIVE_WORDS) neg = len(words & NEGATIVE_WORDS) total = pos + neg if total == 0: return 0.0 return round((pos - neg) / total, 2) def _classify(score: float) -> str: """Classify sentiment score into label.""" if score > 0.15: return "bullish" elif score < -0.15: return "bearish" return "neutral" # ── LLM-based Sentiment (Groq) ────────────────────────────────────────── async def _score_headlines_llm(headlines: List[str]) -> List[Dict[str, Any]]: """ Score headlines using Groq LLM for nuanced financial sentiment. Falls back to rule-based scoring on failure. """ from app.config import get_settings settings = get_settings() if not settings.groq_api_key: return [] try: headlines_text = "\n".join(f"{i+1}. {h}" for i, h in enumerate(headlines[:20])) prompt = f"""Analyze each financial news headline and classify its sentiment. Return ONLY a JSON array of objects with "index" (1-based), "sentiment" ("bullish"/"bearish"/"neutral"), and "score" (-1.0 to 1.0). Headlines: {headlines_text} JSON response:""" async with aiohttp.ClientSession() as session: async with session.post( "https://api.groq.com/openai/v1/chat/completions", headers={ "Authorization": f"Bearer {settings.groq_api_key}", "Content-Type": "application/json", }, json={ "model": "llama-3.3-70b-versatile", "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, "max_tokens": 1000, }, timeout=aiohttp.ClientTimeout(total=15), ) as resp: if resp.status != 200: return [] data = await resp.json() content = data["choices"][0]["message"]["content"] # Extract JSON from response import json # Try to find JSON array in the response start = content.find("[") end = content.rfind("]") + 1 if start >= 0 and end > start: results = json.loads(content[start:end]) return results except Exception as e: logger.warning("LLM sentiment failed: %s", e) return [] async def fetch_news_sentiment(ticker: str, use_llm: bool = True) -> Dict[str, Any]: """Fetch and analyze news headlines for a ticker.""" cache_key = f"sentiment:news:{ticker}" cached = await cache_get(cache_key) if cached: return cached headlines: List[Dict[str, Any]] = [] async with aiohttp.ClientSession() as session: for source, url_template in NEWS_FEEDS.items(): try: url = url_template.format(ticker=ticker) async with session.get(url, timeout=aiohttp.ClientTimeout(total=8)) as resp: if resp.status != 200: continue text = await resp.text() root = ElementTree.fromstring(text) for item in root.iter("item"): title_el = item.find("title") pub_date_el = item.find("pubDate") link_el = item.find("link") if title_el is None or title_el.text is None: continue title = title_el.text.strip() rule_score = _score_headline(title) headlines.append({ "title": title, "source": source, "score": rule_score, "sentiment": _classify(rule_score), "method": "rule_based", "published": pub_date_el.text if pub_date_el is not None else None, "url": link_el.text if link_el is not None else None, }) except Exception as e: logger.warning("Failed to fetch %s news for %s: %s", source, ticker, e) # Try LLM-based scoring for better accuracy llm_results = [] if use_llm and headlines: try: llm_results = await _score_headlines_llm([h["title"] for h in headlines[:20]]) except Exception: pass # Merge LLM results into headlines if llm_results: for lr in llm_results: idx = lr.get("index", 0) - 1 if 0 <= idx < len(headlines): headlines[idx]["ai_score"] = lr.get("score", headlines[idx]["score"]) headlines[idx]["ai_sentiment"] = lr.get("sentiment", headlines[idx]["sentiment"]) headlines[idx]["method"] = "ai_enhanced" # Use AI score as primary if available headlines[idx]["score"] = lr.get("score", headlines[idx]["score"]) headlines[idx]["sentiment"] = lr.get("sentiment", headlines[idx]["sentiment"]) # Compute aggregate scores if headlines: avg_score = round(sum(h["score"] for h in headlines) / len(headlines), 3) bullish_count = sum(1 for h in headlines if h["sentiment"] == "bullish") bearish_count = sum(1 for h in headlines if h["sentiment"] == "bearish") neutral_count = sum(1 for h in headlines if h["sentiment"] == "neutral") ai_enhanced = sum(1 for h in headlines if h.get("method") == "ai_enhanced") else: avg_score = 0.0 bullish_count = bearish_count = neutral_count = ai_enhanced = 0 result = { "ticker": ticker, "headline_count": len(headlines), "avg_score": avg_score, "sentiment": _classify(avg_score), "bullish": bullish_count, "bearish": bearish_count, "neutral": neutral_count, "ai_enhanced_count": ai_enhanced, "headlines": headlines[:20], # Latest 20 "timestamp": datetime.utcnow().isoformat(), } await cache_set(cache_key, result, SENTIMENT_CACHE_TTL) return result async def fetch_multi_sentiment(tickers: List[str]) -> List[Dict[str, Any]]: """Fetch sentiment for multiple tickers.""" import asyncio tasks = [fetch_news_sentiment(t) for t in tickers[:10]] results = await asyncio.gather(*tasks, return_exceptions=True) return [r for r in results if isinstance(r, dict)] # ── Trending / Market Mood ─────────────────────────────────────────────── MARKET_TICKERS = [ # US "SPY", "QQQ", "AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "TSLA", # India "RELIANCE.NS", "TCS.NS", "HDFCBANK.NS", "INFY.NS", # Europe "ASML", "SAP", "SHEL", "AZN", # Asia "9988.HK", "7203.T", ] async def get_market_mood() -> Dict[str, Any]: """Get overall market sentiment from major tickers across all markets.""" cache_key = "sentiment:market_mood" cached = await cache_get(cache_key) if cached: return cached results = await fetch_multi_sentiment(MARKET_TICKERS[:10]) if not results: return {"mood": "neutral", "score": 0, "tickers_analyzed": 0} avg = round(sum(r.get("avg_score", 0) for r in results) / len(results), 3) result = { "mood": _classify(avg), "score": avg, "tickers_analyzed": len(results), "breakdown": [ {"ticker": r["ticker"], "score": r["avg_score"], "sentiment": r["sentiment"]} for r in results ], "timestamp": datetime.utcnow().isoformat(), } await cache_set(cache_key, result, SENTIMENT_CACHE_TTL) return result