Spaces:
Running
Running
| ```python | |
| from typing import Dict, Any, List | |
| import feedparser | |
| from .models_registry import model_registry | |
| CRYPTO_NEWS_RSS = [ | |
| "https://cryptopanic.com/news/rss/", | |
| "https://cointelegraph.com/rss", | |
| "https://news.bitcoin.com/feed/" | |
| ] | |
| async def get_crypto_sentiment(symbol: str = "BTC") -> Dict[str, Any]: | |
| headlines = await fetch_crypto_headlines(symbol) | |
| if not headlines: | |
| return {"score": 0, "confidence": 0, "headlines": []} | |
| sentiment = await model_registry.analyze_sentiment(headlines) | |
| return normalize_sentiment(sentiment, headlines) | |
| async def fetch_crypto_headlines(symbol: str) -> List[str]: | |
| headlines = [] | |
| for url in CRYPTO_NEWS_RSS: | |
| try: | |
| feed = feedparser.parse(url) | |
| for entry in feed.entries: | |
| if symbol.lower() in entry.title.lower() or symbol.lower() in entry.description.lower(): | |
| headlines.append(entry.title) | |
| if len(headlines) >= 10: # Limit to 10 headlines | |
| return headlines | |
| except Exception: | |
| continue | |
| return headlines | |
| def normalize_sentiment(raw_sentiment: Dict[str, Any], headlines: List[str]) -> Dict[str, Any]: | |
| positive = 0 | |
| negative = 0 | |
| neutral = 0 | |
| for item in raw_sentiment.get("sentiment", []): | |
| label = item.get("label", "").lower() | |
| if "positive" in label: | |
| positive += 1 | |
| elif "negative" in label: | |
| negative += 1 | |
| else: | |
| neutral += 1 | |
| total = len(headlines) | |
| if total == 0: | |
| return {"score": 0, "confidence": 0, "headlines": headlines} | |
| score = (positive - negative) / total | |
| confidence = max(positive, negative, neutral) / total | |
| return { | |
| "score": score, | |
| "confidence": confidence, | |
| "bullish": positive / total, | |
| "bearish": negative / total, | |
| "neutral": neutral / total, | |
| "headlines": headlines | |
| } | |
| ``` |