app.py

import gradio as gr
import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import feedparser
from textblob import TextBlob
from statsmodels.tsa.holtwinters import ExponentialSmoothing

# Function to fetch cryptocurrency data
def get_crypto_data(symbol, period="30d", interval="1h"):
    crypto = yf.Ticker(f"{symbol}-USD")
    data = crypto.history(period=period, interval=interval)
    return data

# Function to calculate RSI
def calculate_rsi(data, period=14):
    delta = data['Close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
    rs = gain / loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

# Function to calculate MACD
def calculate_macd(data, short_window=12, long_window=26, signal_window=9):
    short_ema = data['Close'].ewm(span=short_window, adjust=False).mean()
    long_ema = data['Close'].ewm(span=long_window, adjust=False).mean()
    macd = short_ema - long_ema
    signal = macd.ewm(span=signal_window, adjust=False).mean()
    return macd, signal

# Function to calculate EMA
def calculate_ema(data, period=20):
    return data['Close'].ewm(span=period, adjust=False).mean()

# Function to calculate ATR
def calculate_atr(data, period=14):
    high_low = data['High'] - data['Low']
    high_close = np.abs(data['High'] - data['Close'].shift())
    low_close = np.abs(data['Low'] - data['Close'].shift())
    true_range = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
    atr = true_range.rolling(window=period).mean()
    return atr

# Function to calculate Stochastic Oscillator
def calculate_stochastic(data, period=14):
    high = data['High'].rolling(window=period).max()
    low = data['Low'].rolling(window=period).min()
    stoch_k = 100 * ((data['Close'] - low) / (high - low))
    stoch_d = stoch_k.rolling(window=3).mean()
    return stoch_k, stoch_d

# Function to calculate Bollinger Bands
def calculate_bollinger_bands(data, period=20, std_dev=2):
    sma = data['Close'].rolling(window=period).mean()
    std = data['Close'].rolling(window=period).std()
    upper_band = sma + (std * std_dev)
    lower_band = sma - (std * std_dev)
    return upper_band, lower_band

# Function to calculate On-Balance Volume (OBV)
def calculate_obv(data):
    obv = (np.sign(data['Close'].diff()) * data['Volume']).cumsum()
    return obv

# Function to calculate Average Directional Index (ADX)
def calculate_adx(data, period=14):
    high = data['High']
    low = data['Low']
    close = data['Close']
    
    # Calculate +DM and -DM
    plus_dm = high.diff()
    minus_dm = -low.diff()
    plus_dm[plus_dm < 0] = 0
    minus_dm[minus_dm < 0] = 0
    
    # Calculate True Range (TR)
    tr = pd.concat([high - low, abs(high - close.shift()), abs(low - close.shift())], axis=1).max(axis=1)
    
    # Calculate +DI and -DI
    plus_di = 100 * (plus_dm.ewm(alpha=1/period).mean() / tr.ewm(alpha=1/period).mean())
    minus_di = 100 * (minus_dm.ewm(alpha=1/period).mean() / tr.ewm(alpha=1/period).mean())
    
    # Calculate ADX
    dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di)
    adx = dx.ewm(alpha=1/period).mean()
    return adx

# Function to calculate Fibonacci Retracement levels
def calculate_fibonacci_levels(data):
    high = data['High'].max()
    low = data['Low'].min()
    diff = high - low
    return {
        "23.6%": high - diff * 0.236,
        "38.2%": high - diff * 0.382,
        "50%": high - diff * 0.5,
        "61.8%": high - diff * 0.618,
        "78.6%": high - diff * 0.786,
    }

# Function to calculate probabilities
def calculate_probabilities(data):
    # Calculate indicators
    data['RSI'] = calculate_rsi(data)
    data['MACD'], data['MACD_Signal'] = calculate_macd(data)
    data['EMA_50'] = calculate_ema(data, period=50)
    data['EMA_200'] = calculate_ema(data, period=200)
    data['ATR'] = calculate_atr(data)
    data['Stoch_K'], data['Stoch_D'] = calculate_stochastic(data)
    data['Upper_Band'], data['Lower_Band'] = calculate_bollinger_bands(data)
    data['OBV'] = calculate_obv(data)
    data['ADX'] = calculate_adx(data)
    
    # Use the most recent values for predictions
    recent_data = data.iloc[-1]
    
    # Calculate probabilities
    probabilities = {
        "RSI": {"Pump": 0, "Dump": 0},
        "MACD": {"Pump": 0, "Dump": 0},
        "EMA": {"Pump": 0, "Dump": 0},
        "ATR": {"Pump": 0, "Dump": 0},
        "Stochastic": {"Pump": 0, "Dump": 0},
        "Bollinger Bands": {"Pump": 0, "Dump": 0},
        "OBV": {"Pump": 0, "Dump": 0},
        "ADX": {"Pump": 0, "Dump": 0},
    }
    
    # RSI
    rsi = recent_data['RSI']
    if rsi < 25:
        probabilities["RSI"]["Pump"] = 90  # Strong Pump
    elif 25 <= rsi < 30:
        probabilities["RSI"]["Pump"] = 60  # Moderate Pump
    elif 70 < rsi <= 75:
        probabilities["RSI"]["Dump"] = 60  # Moderate Dump
    elif rsi > 75:
        probabilities["RSI"]["Dump"] = 90  # Strong Dump
    
    # MACD
    macd = recent_data['MACD']
    macd_signal = recent_data['MACD_Signal']
    if macd > macd_signal and macd > 0:
        probabilities["MACD"]["Pump"] = 90  # Strong Pump
    elif macd > macd_signal and macd <= 0:
        probabilities["MACD"]["Pump"] = 60  # Moderate Pump
    elif macd < macd_signal and macd >= 0:
        probabilities["MACD"]["Dump"] = 60  # Moderate Dump
    elif macd < macd_signal and macd < 0:
        probabilities["MACD"]["Dump"] = 90  # Strong Dump
    
    # EMA
    ema_short = recent_data['EMA_50']
    ema_long = recent_data['EMA_200']
    close = recent_data['Close']
    if ema_short > ema_long and close > ema_short:
        probabilities["EMA"]["Pump"] = 90  # Strong Pump
    elif ema_short > ema_long and close <= ema_short:
        probabilities["EMA"]["Pump"] = 60  # Moderate Pump
    elif ema_short < ema_long and close >= ema_short:
        probabilities["EMA"]["Dump"] = 60  # Moderate Dump
    elif ema_short < ema_long and close < ema_short:
        probabilities["EMA"]["Dump"] = 90  # Strong Dump
    
    # ATR
    atr = recent_data['ATR']
    if atr > 100:
        probabilities["ATR"]["Pump"] = 90  # Strong Pump
    elif 50 < atr <= 100:
        probabilities["ATR"]["Pump"] = 60  # Moderate Pump
    elif -100 <= atr < -50:
        probabilities["ATR"]["Dump"] = 60  # Moderate Dump
    elif atr < -100:
        probabilities["ATR"]["Dump"] = 90  # Strong Dump
    
    # Stochastic Oscillator
    stoch_k = recent_data['Stoch_K']
    stoch_d = recent_data['Stoch_D']
    if stoch_k < 20 and stoch_d < 20:
        probabilities["Stochastic"]["Pump"] = 90  # Strong Pump
    elif 20 <= stoch_k < 30 and 20 <= stoch_d < 30:
        probabilities["Stochastic"]["Pump"] = 60  # Moderate Pump
    elif 70 < stoch_k <= 80 and 70 < stoch_d <= 80:
        probabilities["Stochastic"]["Dump"] = 60  # Moderate Dump
    elif stoch_k > 80 and stoch_d > 80:
        probabilities["Stochastic"]["Dump"] = 90  # Strong Dump
    
    # Bollinger Bands
    close = recent_data['Close']
    upper_band = recent_data['Upper_Band']
    lower_band = recent_data['Lower_Band']
    if close <= lower_band:
        probabilities["Bollinger Bands"]["Pump"] = 90  # Strong Pump
    elif lower_band < close <= lower_band * 1.05:
        probabilities["Bollinger Bands"]["Pump"] = 60  # Moderate Pump
    elif upper_band * 0.95 <= close < upper_band:
        probabilities["Bollinger Bands"]["Dump"] = 60  # Moderate Dump
    elif close >= upper_band:
        probabilities["Bollinger Bands"]["Dump"] = 90  # Strong Dump
    
    # OBV
    obv = recent_data['OBV']
    if obv > 100000:
        probabilities["OBV"]["Pump"] = 90  # Strong Pump
    elif 50000 < obv <= 100000:
        probabilities["OBV"]["Pump"] = 60  # Moderate Pump
    elif -100000 <= obv < -50000:
        probabilities["OBV"]["Dump"] = 60  # Moderate Dump
    elif obv < -100000:
        probabilities["OBV"]["Dump"] = 90  # Strong Dump
    
    # ADX
    adx = recent_data['ADX']
    if adx > 25:
        probabilities["ADX"]["Pump"] = 90  # Strong Pump
    elif 20 < adx <= 25:
        probabilities["ADX"]["Pump"] = 60  # Moderate Pump
    elif 15 < adx <= 20:
        probabilities["ADX"]["Dump"] = 60  # Moderate Dump
    elif adx <= 15:
        probabilities["ADX"]["Dump"] = 90  # Strong Dump
    
    return probabilities, recent_data

# Function to predict future prices using Exponential Smoothing
def predict_price(data, days=7):
    try:
        # Prepare data for Exponential Smoothing
        df = data[['Close']]
        
        # Train the model
        model = ExponentialSmoothing(df, trend="add", seasonal="add", seasonal_periods=7)
        fit = model.fit()
        
        # Make future predictions
        forecast = fit.forecast(steps=days)
        
        # Format predictions with dates
        last_date = data.index[-1]
        dates = pd.date_range(start=last_date + timedelta(days=1), periods=days)
        forecast_df = pd.DataFrame({"Date": dates, "Price": forecast})
        forecast_df["Date"] = forecast_df["Date"].dt.strftime("%B %d")  # Format as "Month Day"
        forecast_df["Price"] = forecast_df["Price"].round(2)
        
        return forecast_df
    except Exception as e:
        return f"Error predicting prices: {e}"

# Function to fetch news from top 5 crypto news sites and perform sentiment analysis
def fetch_crypto_news(symbol):
    try:
        # List of RSS feeds for top 5 crypto news sites
        rss_feeds = [
            "https://coindesk.com/feed/",
            "https://cointelegraph.com/rss",
            "https://cryptoslate.com/feed/",
            "https://www.newsbtc.com/feed/",
            "https://news.bitcoin.com/feed/"
        ]
        
        news_items = []
        for feed_url in rss_feeds:
            feed = feedparser.parse(feed_url)
            for entry in feed.entries[:5]:  # Limit to 5 articles per site
                if symbol.lower() in entry.title.lower() or symbol.lower() in entry.summary.lower():
                    # Perform sentiment analysis on the article title
                    analysis = TextBlob(entry.title)
                    sentiment = "Bullish" if analysis.sentiment.polarity > 0 else "Bearish" if analysis.sentiment.polarity < 0 else "Neutral"
                    
                    # Format the date as "Month Day"
                    published_date = datetime.strptime(entry.published, "%a, %d %b %Y %H:%M:%S %z").strftime("%B %d")
                    
                    # Extract website name from the link
                    website = entry.link.split("//")[1].split("/")[0]
                    
                    news_items.append({
                        "title": entry.title,
                        "website": website,
                        "sentiment": sentiment,
                        "published": published_date,
                    })
        return news_items[:5]  # Return top 5 articles
    except Exception as e:
        return f"Error fetching crypto news: {e}"

# Gradio Interface
def crypto_app(symbol):
    if symbol:
        # Fetch data
        data = get_crypto_data(symbol)
        if data.empty:
            return f"No data found for {symbol}. Please check the symbol and try again."
        else:
            # Ensure the DataFrame has enough rows
            if len(data) < 20:
                return f"Not enough data to calculate indicators. Only {len(data)} rows available. Please try a longer period."
            else:
                # Calculate probabilities
                probabilities, recent_data = calculate_probabilities(data)
                
                # Predict future prices
                price_predictions = predict_price(data)
                
                # Fetch crypto news and sentiment
                news_items = fetch_crypto_news(symbol)
                
                # Calculate Fibonacci Retracement levels
                fib_levels = calculate_fibonacci_levels(data)
                
                # Prepare output
                output = f"**{symbol} Pump/Dump Probabilities:**\n"
                for indicator, values in probabilities.items():
                    output += f"- **{indicator}**: Pump: {values['Pump']:.2f}%, Dump: {values['Dump']:.2f}%\n"
                
                output += "\n**Price Predictions (Next 7 Days):**\n"
                if isinstance(price_predictions, pd.DataFrame):
                    output += price_predictions.to_string(index=False)
                else:
                    output += price_predictions
                
                output += "\n\n**Fibonacci Retracement Levels:**\n"
                for level, price in fib_levels.items():
                    output += f"- **{level}**: ${price:.2f}\n"
                
                output += "\n**Latest Crypto News Sentiment:**\n"
                if isinstance(news_items, list):
                    for news in news_items:
                        output += f"- **{news['title']}** ({news['sentiment']}) - {news['website']}\n"
                else:
                    output += news_items
                
                return output
    else:
        return "Please enter a cryptocurrency symbol."

# Gradio Interface with Background Image
iface = gr.Interface(
    fn=crypto_app,
    inputs=gr.Textbox(placeholder="Enter cryptocurrency symbol (e.g., ETH, BTC)"),
    outputs="text",
    title="Crypto AI Agent 📈📉",
    description="This app provides technical indicator-based predictions, price forecasts, and sentiment analysis for any cryptocurrency.",
    theme="default",  # Use a theme that supports custom backgrounds
    css=".gradio-container { background-image: url('https://example.com/crypto-background.jpg'); background-size: cover; }",
)

iface.launch()