Spaces:
Running
Running
File size: 1,975 Bytes
da3a00e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | ```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
}
``` |