42 lines
1.5 KiB
Python
42 lines
1.5 KiB
Python
|
import logging
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
POSITIVE_WORDS = {"surge", "rally", "bullish", "high", "gain", "profit", "grows", "adoption", "breakout", "soar", "record"}
|
||
|
|
NEGATIVE_WORDS = {"crash", "drop", "bearish", "low", "loss", "fear", "down", "dump", "scam", "hack", "ban", "panic", "plunge"}
|
||
|
|
|
||
|
|
class NewsAgent:
|
||
|
|
def analyze(self, articles: list[dict]) -> float:
|
||
|
|
if not articles:
|
||
|
|
return 50.0
|
||
|
|
try:
|
||
|
|
scores = []
|
||
|
|
for article in articles:
|
||
|
|
vote_score = self._vote_sentiment(article.get("sentiment_votes", {}))
|
||
|
|
text_score = self._text_sentiment(article.get("title", ""))
|
||
|
|
if vote_score is not None:
|
||
|
|
scores.append(vote_score * 0.6 + text_score * 0.4)
|
||
|
|
else:
|
||
|
|
scores.append(text_score)
|
||
|
|
return round(max(0, min(100, sum(scores) / len(scores))), 1)
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"News analysis error: {e}")
|
||
|
|
return 50.0
|
||
|
|
|
||
|
|
def _vote_sentiment(self, votes: dict) -> float | None:
|
||
|
|
pos = votes.get("positive", 0)
|
||
|
|
neg = votes.get("negative", 0)
|
||
|
|
total = pos + neg
|
||
|
|
if total == 0:
|
||
|
|
return None
|
||
|
|
return (pos / total) * 100
|
||
|
|
|
||
|
|
def _text_sentiment(self, text: str) -> float:
|
||
|
|
words = set(text.lower().split())
|
||
|
|
pos = len(words & POSITIVE_WORDS)
|
||
|
|
neg = len(words & NEGATIVE_WORDS)
|
||
|
|
total = pos + neg
|
||
|
|
if total == 0:
|
||
|
|
return 50.0
|
||
|
|
return (pos / total) * 100
|