api_fallback_demo.py

#!/usr/bin/env python
"""
API Fallback Demonstration Script

This script demonstrates the resilient API client with:
1. CoinMarketCap fallback for CoinGecko
2. Multi-source social sentiment data integration
3. Automatic normalization of data across providers

Usage:
    python api_fallback_demo.py [symbol]
    
    Example: python api_fallback_demo.py BTC
"""

import os
import sys
import json
import logging
import asyncio
import argparse
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("api_fallback_demo")

# Import the resilient API client
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from src.collectors.resilient_api_client import ResilientApiClient, COIN_MAPPING

async def test_price_fallback(client, symbol):
    """Test price data with fallback capability."""
    print(f"\n---- TESTING PRICE DATA FOR {symbol} ----")
    print("First try to get data from CoinGecko, with fallback to CoinMarketCap:")
    
    price_data = await client.get_price_data(symbol)
    
    print(f"\n📊 {symbol} Price Data:")
    print(f"Sources used: {', '.join(price_data.sources)}")
    
    if not price_data.sources:
        print("❌ No price data available from any source")
        return
    
    # Print price data
    if price_data.price_usd is not None:
        print(f"Price: ${price_data.price_usd:,.2f}")
    else:
        print("Price: Not available")
        
    if price_data.market_cap_usd is not None:
        print(f"Market Cap: ${price_data.market_cap_usd:,.0f}")
    else:
        print("Market Cap: Not available")
        
    if price_data.volume_24h_usd is not None:
        print(f"24h Volume: ${price_data.volume_24h_usd:,.0f}")
    else:
        print("24h Volume: Not available")
        
    if price_data.change_24h_percent is not None:
        print(f"24h Change: {price_data.change_24h_percent:.2f}%")
    else:
        print("24h Change: Not available")
    
    # Show fallback status
    if "coingecko" in price_data.sources:
        print("\nℹ️ CoinGecko was available - no fallback needed")
    elif "coinmarketcap" in price_data.sources:
        print("\n✅ FALLBACK SUCCESSFUL: Used CoinMarketCap when CoinGecko failed")
    
    return price_data

async def test_social_integration(client, symbol):
    """Test social data integration from multiple sources."""
    print(f"\n---- TESTING SOCIAL DATA INTEGRATION FOR {symbol} ----")
    print("Collecting and normalizing social data from multiple sources:")
    
    social_data = await client.get_social_data(symbol)
    
    print(f"\n🔍 {symbol} Social Data:")
    print(f"Sources used: {', '.join(social_data.sources)}")
    
    if not social_data.sources:
        print("❌ No social data available from any source")
        return
    
    # Print social metrics
    if social_data.twitter_followers is not None:
        print(f"Twitter Followers: {social_data.twitter_followers:,}")
    else:
        print("Twitter Followers: Not available")
        
    if social_data.reddit_subscribers is not None:
        print(f"Reddit Subscribers: {social_data.reddit_subscribers:,}")
    else:
        print("Reddit Subscribers: Not available")
        
    if social_data.reddit_active_users is not None:
        print(f"Reddit Active Users: {social_data.reddit_active_users:,}")
    else:
        print("Reddit Active Users: Not available")
    
    # Print sentiment score if available
    if social_data.sentiment_score is not None:
        print(f"Calculated Sentiment Score: {social_data.sentiment_score:.1f}/100")
        
        # Show sentiment classification
        if social_data.sentiment_score > 80:
            print("Classification: Very Positive")
        elif social_data.sentiment_score > 60:
            print("Classification: Positive")
        elif social_data.sentiment_score > 40:
            print("Classification: Neutral")
        elif social_data.sentiment_score > 20:
            print("Classification: Negative")
        else:
            print("Classification: Very Negative")
    else:
        print("Sentiment Score: Not available")
    
    # Show data integration success
    if len(social_data.sources) > 1:
        print(f"\n✅ INTEGRATION SUCCESSFUL: Combined data from {len(social_data.sources)} sources")
    else:
        print(f"\nℹ️ Used data from only {social_data.sources[0]}")
    
    return social_data

async def display_available_symbols():
    """Display available cryptocurrency symbols."""
    print("\nAvailable cryptocurrency symbols:")
    for symbol in sorted(COIN_MAPPING.keys()):
        print(f"- {symbol}")

async def main():
    """Main function to demonstrate API fallback and data integration."""
    # Parse command line arguments
    parser = argparse.ArgumentParser(description="Demonstrate API fallback and data integration")
    parser.add_argument("symbol", nargs="?", default="BTC", help="Cryptocurrency symbol to test")
    args = parser.parse_args()
    
    symbol = args.symbol.upper()
    
    # Check if the symbol is supported
    if symbol not in COIN_MAPPING:
        print(f"❌ Symbol {symbol} is not supported")
        await display_available_symbols()
        return
    
    print(f"🚀 Testing API fallback and data integration for {symbol}")
    
    async with ResilientApiClient() as client:
        # Test price data with fallback
        await test_price_fallback(client, symbol)
        
        # Test social data integration
        await test_social_integration(client, symbol)
    
    print("\n✅ Demo completed!")

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