From b9d21e2e2f7ae096c2f8a01bb142a685683b5b90 Mon Sep 17 00:00:00 2001 From: TheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com> Date: Thu, 2 Apr 2026 09:44:43 +0900 Subject: feat: add market sentiment filters (Fear & Greed, CryptoPanic, CryptoQuant) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - SentimentProvider: fetches Fear & Greed Index (free, no key), CryptoPanic news sentiment (free key), CryptoQuant exchange netflow (free key) - SentimentData: aggregated should_buy/should_block logic - Fear < 30 = buy opportunity, Greed > 80 = block buying - Negative news < -0.5 = block buying - Exchange outflow = bullish, inflow = bearish - Integrated into Asian Session RSI strategy as entry filter - All providers optional — disabled when API key missing - 14 sentiment tests + 386 total tests passing --- shared/src/shared/config.py | 2 + shared/src/shared/sentiment.py | 219 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 221 insertions(+) create mode 100644 shared/src/shared/sentiment.py (limited to 'shared/src') diff --git a/shared/src/shared/config.py b/shared/src/shared/config.py index 6023755..def70b9 100644 --- a/shared/src/shared/config.py +++ b/shared/src/shared/config.py @@ -36,5 +36,7 @@ class Settings(BaseSettings): circuit_breaker_threshold: int = 5 circuit_breaker_timeout: int = 60 metrics_auth_token: str = "" # If set, /health and /metrics require Bearer token + cryptopanic_api_key: str = "" # Free key from cryptopanic.com + cryptoquant_api_key: str = "" # Free key from cryptoquant.com model_config = {"env_file": ".env", "env_file_encoding": "utf-8"} diff --git a/shared/src/shared/sentiment.py b/shared/src/shared/sentiment.py new file mode 100644 index 0000000..bc62efe --- /dev/null +++ b/shared/src/shared/sentiment.py @@ -0,0 +1,219 @@ +"""Market sentiment data from free APIs. + +Supports: +- Fear & Greed Index (alternative.me) — no API key needed +- CryptoPanic news sentiment — free API key from cryptopanic.com +- CryptoQuant exchange netflow — free API key from cryptoquant.com + +All providers are optional. If API key is missing, the provider is disabled. +""" + +import logging +from dataclasses import dataclass, field +from datetime import datetime, timezone + +import aiohttp + +logger = logging.getLogger(__name__) + + +@dataclass +class SentimentData: + """Aggregated sentiment snapshot.""" + + fear_greed_value: int | None = None # 0-100 + fear_greed_label: str | None = ( + None # "Extreme Fear", "Fear", "Neutral", "Greed", "Extreme Greed" + ) + news_sentiment: float | None = None # -1.0 (bearish) to 1.0 (bullish) + news_count: int = 0 + exchange_netflow: float | None = ( + None # Positive = inflow (bearish), Negative = outflow (bullish) + ) + timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + + @property + def should_buy(self) -> bool: + """Simple aggregated buy signal from sentiment.""" + score = 0 + checks = 0 + + if self.fear_greed_value is not None: + checks += 1 + if self.fear_greed_value < 30: + score += 1 # Fear = buy opportunity + elif self.fear_greed_value > 70: + score -= 1 # Greed = avoid buying + + if self.news_sentiment is not None: + checks += 1 + if self.news_sentiment > 0.1: + score += 1 # Positive news + elif self.news_sentiment < -0.3: + score -= 1 # Very negative news = avoid + + if self.exchange_netflow is not None: + checks += 1 + if self.exchange_netflow < 0: + score += 1 # Outflow = bullish (coins leaving exchanges) + elif self.exchange_netflow > 0: + score -= 1 # Inflow = bearish (coins entering exchanges to sell) + + if checks == 0: + return True # No data, allow by default + + return score >= 0 # Net neutral or positive = allow + + @property + def should_block(self) -> bool: + """Strong bearish signal — block all buying.""" + # Block on extreme greed + if self.fear_greed_value is not None and self.fear_greed_value > 80: + return True + # Block on very negative news + if self.news_sentiment is not None and self.news_sentiment < -0.5: + return True + return False + + +class SentimentProvider: + """Fetches sentiment data from multiple free APIs.""" + + def __init__( + self, + cryptopanic_api_key: str = "", + cryptoquant_api_key: str = "", + ) -> None: + self._cryptopanic_key = cryptopanic_api_key + self._cryptoquant_key = cryptoquant_api_key + self._session: aiohttp.ClientSession | None = None + self._cached: SentimentData | None = None + self._cache_ttl: int = 300 # 5 minutes cache + + async def _ensure_session(self) -> aiohttp.ClientSession: + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession() + return self._session + + async def fetch_fear_greed(self) -> tuple[int | None, str | None]: + """Fetch Fear & Greed Index from alternative.me (no key needed).""" + try: + session = await self._ensure_session() + url = "https://api.alternative.me/fng/?limit=1" + async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp: + if resp.status != 200: + logger.warning("Fear & Greed API returned %d", resp.status) + return None, None + data = await resp.json() + entry = data.get("data", [{}])[0] + value = int(entry.get("value", 0)) + label = entry.get("value_classification", "") + return value, label + except Exception as exc: + logger.warning("Fear & Greed fetch failed: %s", exc) + return None, None + + async def fetch_news_sentiment(self, currency: str = "SOL") -> tuple[float | None, int]: + """Fetch news sentiment from CryptoPanic. + + Returns (sentiment_score, news_count). + Sentiment score: -1.0 (all bearish) to 1.0 (all bullish). + """ + if not self._cryptopanic_key: + return None, 0 + + try: + session = await self._ensure_session() + url = ( + f"https://cryptopanic.com/api/v1/posts/" + f"?auth_token={self._cryptopanic_key}" + f"¤cies={currency}" + f"&kind=news" + f"&filter=hot" + f"&limit=10" + ) + async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp: + if resp.status != 200: + logger.warning("CryptoPanic API returned %d", resp.status) + return None, 0 + data = await resp.json() + results = data.get("results", []) + if not results: + return None, 0 + + # CryptoPanic provides votes: positive, negative, important, etc. + total_positive = 0 + total_negative = 0 + count = 0 + for post in results: + votes = post.get("votes", {}) + pos = votes.get("positive", 0) + neg = votes.get("negative", 0) + total_positive += pos + total_negative += neg + count += 1 + + total_votes = total_positive + total_negative + if total_votes == 0: + return 0.0, count + + sentiment = (total_positive - total_negative) / total_votes + return sentiment, count + except Exception as exc: + logger.warning("CryptoPanic fetch failed: %s", exc) + return None, 0 + + async def fetch_exchange_netflow(self, symbol: str = "sol") -> float | None: + """Fetch exchange netflow from CryptoQuant. + + Returns netflow value. Positive = inflow (bearish), Negative = outflow (bullish). + """ + if not self._cryptoquant_key: + return None + + try: + session = await self._ensure_session() + url = ( + f"https://api.cryptoquant.com/v1/{symbol}/exchange-flows/netflow?window=day&limit=1" + ) + headers = {"Authorization": f"Bearer {self._cryptoquant_key}"} + async with session.get( + url, headers=headers, timeout=aiohttp.ClientTimeout(total=10) + ) as resp: + if resp.status != 200: + logger.warning("CryptoQuant API returned %d", resp.status) + return None + data = await resp.json() + result = data.get("result", {}).get("data", []) + if result: + return float(result[0].get("netflow", 0)) + return None + except Exception as exc: + logger.warning("CryptoQuant fetch failed: %s", exc) + return None + + async def get_sentiment(self, currency: str = "SOL") -> SentimentData: + """Fetch all sentiment data and return aggregated result.""" + fg_value, fg_label = await self.fetch_fear_greed() + news_score, news_count = await self.fetch_news_sentiment(currency) + netflow = await self.fetch_exchange_netflow(currency.lower()) + + sentiment = SentimentData( + fear_greed_value=fg_value, + fear_greed_label=fg_label, + news_sentiment=news_score, + news_count=news_count, + exchange_netflow=netflow, + ) + + self._cached = sentiment + return sentiment + + @property + def cached(self) -> SentimentData | None: + """Return last fetched sentiment data.""" + return self._cached + + async def close(self) -> None: + if self._session and not self._session.closed: + await self._session.close() -- cgit v1.2.3