feat: technical, news, and social analysis agents

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

agents/news.py

@@ -0,0 +1,41 @@
+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

agents/social.py

@@ -0,0 +1,33 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+class SocialAgent:
+    def analyze(self, sentiment: dict, mention_trend: float = 1.0) -> float:
+        total = sentiment.get("total", 0)
+        if total == 0:
+            return 50.0
+        try:
+            pos = sentiment.get("positive", 0)
+            ratio = pos / total
+            sentiment_score = ratio * 100
+
+            if mention_trend > 2.0:
+                trend_bonus = 15
+            elif mention_trend > 1.5:
+                trend_bonus = 10
+            elif mention_trend > 1.0:
+                trend_bonus = 5
+            elif mention_trend < 0.5:
+                trend_bonus = -10
+            else:
+                trend_bonus = 0
+
+            if ratio < 0.4 and mention_trend > 1.5:
+                trend_bonus = -15
+
+            score = sentiment_score + trend_bonus
+            return round(max(0, min(100, score)), 1)
+        except Exception as e:
+            logger.error(f"Social analysis error: {e}")
+            return 50.0

agents/technical.py

@@ -0,0 +1,113 @@
+import pandas as pd
+import ta
+import logging
+
+logger = logging.getLogger(__name__)
+
+class TechnicalAgent:
+    MIN_CANDLES = 30
+
+    def analyze(self, df: pd.DataFrame) -> float:
+        if len(df) < self.MIN_CANDLES:
+            return 50.0
+
+        try:
+            signals = []
+
+            # RSI (14)
+            rsi = ta.momentum.RSIIndicator(df["close"], window=14).rsi().iloc[-1]
+            if rsi < 30:
+                signals.append(65)
+            elif rsi < 45:
+                signals.append(70)
+            elif rsi < 55:
+                signals.append(50)
+            elif rsi < 70:
+                signals.append(35)
+            else:
+                signals.append(40)
+
+            # MACD
+            macd_ind = ta.trend.MACD(df["close"])
+            macd_diff = macd_ind.macd_diff().iloc[-1]
+            prev_diff = macd_ind.macd_diff().iloc[-2]
+            if macd_diff > 0 and prev_diff <= 0:
+                signals.append(85)
+            elif macd_diff < 0 and prev_diff >= 0:
+                signals.append(15)
+            elif macd_diff > 0:
+                signals.append(65)
+            else:
+                signals.append(35)
+
+            # Bollinger Bands
+            bb = ta.volatility.BollingerBands(df["close"])
+            bb_high = bb.bollinger_hband().iloc[-1]
+            bb_low = bb.bollinger_lband().iloc[-1]
+            price = df["close"].iloc[-1]
+            bb_pct = (price - bb_low) / (bb_high - bb_low) if (bb_high - bb_low) > 0 else 0.5
+            if bb_pct < 0.2:
+                signals.append(55)
+            elif bb_pct > 0.8:
+                signals.append(40)
+            else:
+                signals.append(50)
+
+            # SMA trend (20 vs 50)
+            sma20 = df["close"].rolling(20).mean().iloc[-1]
+            sma50 = df["close"].rolling(50).mean().iloc[-1] if len(df) >= 50 else sma20
+            if price > sma20 > sma50:
+                signals.append(80)
+            elif price > sma20:
+                signals.append(65)
+            elif price < sma20 < sma50:
+                signals.append(20)
+            else:
+                signals.append(40)
+
+            # SMA 200 (long-term trend)
+            if len(df) >= 200:
+                sma200 = df["close"].rolling(200).mean().iloc[-1]
+                if price > sma200:
+                    signals.append(70)
+                else:
+                    signals.append(30)
+
+            # Volume trend
+            vol_avg = df["volume"].rolling(20).mean().iloc[-1]
+            vol_current = df["volume"].iloc[-1]
+            vol_ratio = vol_current / vol_avg if vol_avg > 0 else 1.0
+            if vol_ratio > 2.0 and price > sma20:
+                signals.append(80)
+            elif vol_ratio > 2.0:
+                signals.append(40)
+            else:
+                signals.append(50)
+
+            # OBV trend
+            obv = ta.volume.OnBalanceVolumeIndicator(df["close"], df["volume"]).on_balance_volume()
+            obv_sma = obv.rolling(20).mean()
+            if obv.iloc[-1] > obv_sma.iloc[-1]:
+                signals.append(65)
+            else:
+                signals.append(35)
+
+            # Candlestick patterns (simplified)
+            last = df.iloc[-1]
+            prev = df.iloc[-2]
+            body = last["close"] - last["open"]
+            prev_body = prev["close"] - prev["open"]
+            if prev_body < 0 and body > 0 and body > abs(prev_body):
+                signals.append(80)  # bullish engulfing
+            elif prev_body > 0 and body < 0 and abs(body) > prev_body:
+                signals.append(20)  # bearish engulfing
+            elif abs(body) < (last["high"] - last["low"]) * 0.1:
+                signals.append(50)  # doji
+            else:
+                signals.append(50)
+
+            return round(sum(signals) / len(signals), 1)
+
+        except Exception as e:
+            logger.error(f"Technical analysis error: {e}")
+            return 50.0

tests/test_news_agent.py

@@ -0,0 +1,31 @@
+import pytest
+from agents.news import NewsAgent
+
+def test_score_positive_news():
+    agent = NewsAgent()
+    articles = [
+        {"title": "Bitcoin surges 10%", "sentiment_votes": {"positive": 10, "negative": 1}},
+        {"title": "BTC adoption grows", "sentiment_votes": {"positive": 8, "negative": 2}},
+        {"title": "Bitcoin rally continues", "sentiment_votes": {"positive": 15, "negative": 0}},
+    ]
+    score = agent.analyze(articles)
+    assert score >= 70
+
+def test_score_negative_news():
+    agent = NewsAgent()
+    articles = [
+        {"title": "Bitcoin crashes hard", "sentiment_votes": {"positive": 1, "negative": 10}},
+        {"title": "Crypto market in fear", "sentiment_votes": {"positive": 0, "negative": 15}},
+    ]
+    score = agent.analyze(articles)
+    assert score <= 35
+
+def test_score_no_news_returns_50():
+    agent = NewsAgent()
+    assert agent.analyze([]) == 50
+
+def test_score_mixed_news():
+    agent = NewsAgent()
+    articles = [{"title": "BTC up", "sentiment_votes": {"positive": 5, "negative": 5}}]
+    score = agent.analyze(articles)
+    assert 40 <= score <= 60

tests/test_social_agent.py

@@ -0,0 +1,19 @@
+import pytest
+from agents.social import SocialAgent
+
+def test_score_bullish_social():
+    agent = SocialAgent()
+    sentiment = {"positive": 8, "negative": 1, "neutral": 1, "total": 10}
+    score = agent.analyze(sentiment, mention_trend=2.5)
+    assert score >= 70
+
+def test_score_bearish_social():
+    agent = SocialAgent()
+    sentiment = {"positive": 1, "negative": 8, "neutral": 1, "total": 10}
+    score = agent.analyze(sentiment, mention_trend=0.5)
+    assert score <= 35
+
+def test_no_data_returns_50():
+    agent = SocialAgent()
+    score = agent.analyze({"positive": 0, "negative": 0, "neutral": 0, "total": 0}, 1.0)
+    assert score == 50

tests/test_technical.py

@@ -0,0 +1,58 @@
+import pytest
+import pandas as pd
+import numpy as np
+from agents.technical import TechnicalAgent
+
+def make_ohlcv(n=100, base_price=100.0):
+    np.random.seed(42)
+    closes = base_price + np.cumsum(np.random.randn(n) * 2)
+    df = pd.DataFrame({
+        "timestamp": pd.date_range("2026-01-01", periods=n, freq="1h"),
+        "open": closes - np.random.rand(n),
+        "high": closes + np.abs(np.random.randn(n) * 2),
+        "low": closes - np.abs(np.random.randn(n) * 2),
+        "close": closes,
+        "volume": np.random.randint(100, 10000, n).astype(float),
+        "quote_volume": np.random.randint(100000, 1000000, n).astype(float),
+    })
+    return df
+
+def test_score_returns_0_to_100():
+    agent = TechnicalAgent()
+    df = make_ohlcv(100)
+    score = agent.analyze(df)
+    assert 0 <= score <= 100
+
+def test_score_with_uptrend():
+    agent = TechnicalAgent()
+    n = 100
+    closes = np.linspace(100, 200, n)
+    df = pd.DataFrame({
+        "timestamp": pd.date_range("2026-01-01", periods=n, freq="1h"),
+        "open": closes - 1, "high": closes + 2,
+        "low": closes - 2, "close": closes,
+        "volume": np.full(n, 5000.0),
+        "quote_volume": np.full(n, 500000.0),
+    })
+    score = agent.analyze(df)
+    assert score >= 55  # uptrend should score above average
+
+def test_score_with_downtrend():
+    agent = TechnicalAgent()
+    n = 100
+    closes = np.linspace(200, 100, n)
+    df = pd.DataFrame({
+        "timestamp": pd.date_range("2026-01-01", periods=n, freq="1h"),
+        "open": closes + 1, "high": closes + 2,
+        "low": closes - 2, "close": closes,
+        "volume": np.full(n, 5000.0),
+        "quote_volume": np.full(n, 500000.0),
+    })
+    score = agent.analyze(df)
+    assert score <= 45  # downtrend should score below average
+
+def test_insufficient_data_returns_50():
+    agent = TechnicalAgent()
+    df = make_ohlcv(5)
+    score = agent.analyze(df)
+    assert score == 50