crypto_data_bank/collectors/sentiment_collector.py

#!/usr/bin/env python3
"""
جمع‌آوری احساسات بازار از منابع رایگان
Free Market Sentiment Collectors - NO API KEY
"""

import asyncio
import httpx
from typing import Dict, Optional
from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class SentimentCollector:
    """جمع‌آوری احساسات بازار از منابع رایگان"""

    def __init__(self):
        self.timeout = httpx.Timeout(15.0)
        self.headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
            "Accept": "application/json"
        }

    async def collect_fear_greed_index(self) -> Optional[Dict]:
        """
        Alternative.me Crypto Fear & Greed Index
        FREE - No API key needed
        """
        try:
            url = "https://api.alternative.me/fng/"

            async with httpx.AsyncClient(timeout=self.timeout) as client:
                response = await client.get(url, headers=self.headers)

                if response.status_code == 200:
                    data = response.json()

                    if "data" in data and data["data"]:
                        fng = data["data"][0]

                        result = {
                            "fear_greed_value": int(fng.get("value", 50)),
                            "fear_greed_classification": fng.get("value_classification", "Neutral"),
                            "timestamp_fng": fng.get("timestamp"),
                            "source": "alternative.me",
                            "timestamp": datetime.now().isoformat()
                        }

                        logger.info(f"✅ Fear & Greed: {result['fear_greed_value']} ({result['fear_greed_classification']})")
                        return result
                    else:
                        logger.warning("⚠️ Fear & Greed API returned no data")
                        return None
                else:
                    logger.warning(f"⚠️ Fear & Greed returned status {response.status_code}")
                    return None

        except Exception as e:
            logger.error(f"❌ Fear & Greed error: {e}")
            return None

    async def collect_bitcoin_dominance(self) -> Optional[Dict]:
        """
        Bitcoin Dominance from CoinCap
        FREE - No API key needed
        """
        try:
            url = "https://api.coincap.io/v2/assets"
            params = {"limit": 10}

            async with httpx.AsyncClient(timeout=self.timeout) as client:
                response = await client.get(url, params=params, headers=self.headers)

                if response.status_code == 200:
                    data = response.json()
                    assets = data.get("data", [])

                    if not assets:
                        return None

                    # Calculate total market cap
                    total_market_cap = sum(
                        float(asset.get("marketCapUsd", 0))
                        for asset in assets
                        if asset.get("marketCapUsd")
                    )

                    # Get Bitcoin market cap
                    btc = next((a for a in assets if a["symbol"] == "BTC"), None)
                    if not btc:
                        return None

                    btc_market_cap = float(btc.get("marketCapUsd", 0))

                    # Calculate dominance
                    btc_dominance = (btc_market_cap / total_market_cap * 100) if total_market_cap > 0 else 0

                    result = {
                        "btc_dominance": round(btc_dominance, 2),
                        "btc_market_cap": btc_market_cap,
                        "total_market_cap": total_market_cap,
                        "source": "coincap.io",
                        "timestamp": datetime.now().isoformat()
                    }

                    logger.info(f"✅ BTC Dominance: {result['btc_dominance']}%")
                    return result
                else:
                    logger.warning(f"⚠️ CoinCap returned status {response.status_code}")
                    return None

        except Exception as e:
            logger.error(f"❌ BTC Dominance error: {e}")
            return None

    async def collect_global_market_stats(self) -> Optional[Dict]:
        """
        Global Market Statistics from CoinGecko
        FREE - No API key for this endpoint
        """
        try:
            url = "https://api.coingecko.com/api/v3/global"

            async with httpx.AsyncClient(timeout=self.timeout) as client:
                response = await client.get(url, headers=self.headers)

                if response.status_code == 200:
                    data = response.json()
                    global_data = data.get("data", {})

                    if not global_data:
                        return None

                    result = {
                        "total_market_cap_usd": global_data.get("total_market_cap", {}).get("usd", 0),
                        "total_volume_24h_usd": global_data.get("total_volume", {}).get("usd", 0),
                        "btc_dominance": global_data.get("market_cap_percentage", {}).get("btc", 0),
                        "eth_dominance": global_data.get("market_cap_percentage", {}).get("eth", 0),
                        "active_cryptocurrencies": global_data.get("active_cryptocurrencies", 0),
                        "markets": global_data.get("markets", 0),
                        "market_cap_change_24h": global_data.get("market_cap_change_percentage_24h_usd", 0),
                        "source": "coingecko.com",
                        "timestamp": datetime.now().isoformat()
                    }

                    logger.info(f"✅ Global Stats: ${result['total_market_cap_usd']:,.0f} market cap")
                    return result
                else:
                    logger.warning(f"⚠️ CoinGecko global returned status {response.status_code}")
                    return None

        except Exception as e:
            logger.error(f"❌ Global Stats error: {e}")
            return None

    async def calculate_market_sentiment(
        self,
        fear_greed: Optional[Dict],
        btc_dominance: Optional[Dict],
        global_stats: Optional[Dict]
    ) -> Dict:
        """
        محاسبه احساسات کلی بازار
        Calculate overall market sentiment from multiple indicators
        """
        sentiment_score = 50  # Neutral default
        confidence = 0.0
        indicators_count = 0

        sentiment_signals = []

        # Fear & Greed contribution (40% weight)
        if fear_greed:
            fg_value = fear_greed.get("fear_greed_value", 50)
            sentiment_score += (fg_value - 50) * 0.4
            confidence += 0.4
            indicators_count += 1

            sentiment_signals.append({
                "indicator": "fear_greed",
                "value": fg_value,
                "signal": fear_greed.get("fear_greed_classification")
            })

        # BTC Dominance contribution (30% weight)
        if btc_dominance:
            dom_value = btc_dominance.get("btc_dominance", 45)

            # Higher BTC dominance = more fearful (people moving to "safe" crypto)
            # Lower BTC dominance = more greedy (people buying altcoins)
            dom_score = 100 - dom_value  # Inverse relationship
            sentiment_score += (dom_score - 50) * 0.3
            confidence += 0.3
            indicators_count += 1

            sentiment_signals.append({
                "indicator": "btc_dominance",
                "value": dom_value,
                "signal": "Defensive" if dom_value > 50 else "Risk-On"
            })

        # Market Cap Change contribution (30% weight)
        if global_stats:
            mc_change = global_stats.get("market_cap_change_24h", 0)

            # Positive change = bullish, negative = bearish
            mc_score = 50 + (mc_change * 5)  # Scale: -10% change = 0, +10% = 100
            mc_score = max(0, min(100, mc_score))  # Clamp to 0-100

            sentiment_score += (mc_score - 50) * 0.3
            confidence += 0.3
            indicators_count += 1

            sentiment_signals.append({
                "indicator": "market_cap_change_24h",
                "value": mc_change,
                "signal": "Bullish" if mc_change > 0 else "Bearish"
            })

        # Normalize sentiment score to 0-100
        sentiment_score = max(0, min(100, sentiment_score))

        # Determine overall classification
        if sentiment_score >= 75:
            classification = "Extreme Greed"
        elif sentiment_score >= 60:
            classification = "Greed"
        elif sentiment_score >= 45:
            classification = "Neutral"
        elif sentiment_score >= 25:
            classification = "Fear"
        else:
            classification = "Extreme Fear"

        return {
            "overall_sentiment": classification,
            "sentiment_score": round(sentiment_score, 2),
            "confidence": round(confidence, 2),
            "indicators_used": indicators_count,
            "signals": sentiment_signals,
            "fear_greed_value": fear_greed.get("fear_greed_value") if fear_greed else None,
            "fear_greed_classification": fear_greed.get("fear_greed_classification") if fear_greed else None,
            "btc_dominance": btc_dominance.get("btc_dominance") if btc_dominance else None,
            "market_cap_change_24h": global_stats.get("market_cap_change_24h") if global_stats else None,
            "source": "aggregated",
            "timestamp": datetime.now().isoformat()
        }

    async def collect_all_sentiment_data(self) -> Dict:
        """
        جمع‌آوری همه داده‌های احساسات
        Collect ALL sentiment data and calculate overall sentiment
        """
        logger.info("🚀 Starting collection of sentiment data...")

        # Collect all data in parallel
        fear_greed, btc_dom, global_stats = await asyncio.gather(
            self.collect_fear_greed_index(),
            self.collect_bitcoin_dominance(),
            self.collect_global_market_stats(),
            return_exceptions=True
        )

        # Handle exceptions
        fear_greed = fear_greed if not isinstance(fear_greed, Exception) else None
        btc_dom = btc_dom if not isinstance(btc_dom, Exception) else None
        global_stats = global_stats if not isinstance(global_stats, Exception) else None

        # Calculate overall sentiment
        overall_sentiment = await self.calculate_market_sentiment(
            fear_greed,
            btc_dom,
            global_stats
        )

        return {
            "fear_greed": fear_greed,
            "btc_dominance": btc_dom,
            "global_stats": global_stats,
            "overall_sentiment": overall_sentiment
        }


async def main():
    """Test the sentiment collectors"""
    collector = SentimentCollector()

    print("\n" + "="*70)
    print("🧪 Testing FREE Sentiment Collectors")
    print("="*70)

    # Test individual collectors
    print("\n1️⃣ Testing Fear & Greed Index...")
    fg = await collector.collect_fear_greed_index()
    if fg:
        print(f"   Value: {fg['fear_greed_value']}/100")
        print(f"   Classification: {fg['fear_greed_classification']}")

    print("\n2️⃣ Testing Bitcoin Dominance...")
    btc_dom = await collector.collect_bitcoin_dominance()
    if btc_dom:
        print(f"   BTC Dominance: {btc_dom['btc_dominance']}%")
        print(f"   BTC Market Cap: ${btc_dom['btc_market_cap']:,.0f}")

    print("\n3️⃣ Testing Global Market Stats...")
    global_stats = await collector.collect_global_market_stats()
    if global_stats:
        print(f"   Total Market Cap: ${global_stats['total_market_cap_usd']:,.0f}")
        print(f"   24h Volume: ${global_stats['total_volume_24h_usd']:,.0f}")
        print(f"   24h Change: {global_stats['market_cap_change_24h']:.2f}%")

    # Test comprehensive sentiment
    print("\n\n" + "="*70)
    print("📊 Testing Comprehensive Sentiment Analysis")
    print("="*70)

    all_data = await collector.collect_all_sentiment_data()

    overall = all_data["overall_sentiment"]
    print(f"\n✅ Overall Market Sentiment: {overall['overall_sentiment']}")
    print(f"   Sentiment Score: {overall['sentiment_score']}/100")
    print(f"   Confidence: {overall['confidence']:.0%}")
    print(f"   Indicators Used: {overall['indicators_used']}")

    print("\n📊 Individual Signals:")
    for signal in overall.get("signals", []):
        print(f"   • {signal['indicator']}: {signal['value']} ({signal['signal']})")


if __name__ == "__main__":
    asyncio.run(main())