Update app.py

app.py

@@ -1,377 +1,172 @@
-# app.py
 import gradio as gr
 import pandas as pd
 import numpy as np
-from prophet import Prophet
-import plotly.graph_objs as go
-import math
-import requests
-from sklearn.ensemble import RandomForestClassifier
-from textblob import TextBlob
-import yfinance as yf  # Import yfinance
+from transformers import pipeline  # For sentiment analysis
+from sklearn.ensemble import IsolationForest  # For anomaly detection
+import yfinance as yf  # For stock market data
+import requests  # For API calls
 
-# --- Constants ---
-CRYPTO_SYMBOLS = ["BTC-USD", "ETH-USD"]  # Use standardized symbols
-STOCK_SYMBOLS = ["AAPL", "MSFT"]
-INTERVAL_OPTIONS = ["1h", "1d"]  # 1h not available for yfinance for stocks; use 1d for stocks.
+# Replace with your actual API keys
+NEWS_API_KEY = "YOUR_NEWS_API_KEY"  # Get from https://newsapi.org/
+TWITTER_BEARER_TOKEN = "YOUR_TWITTER_BEARER_TOKEN"  # Get from https://developer.twitter.com/
 
-# --- Data Fetching Functions (Revised) ---
-def fetch_crypto_data(symbol, interval="1h", limit=100):
-    """Fetch crypto market data using yfinance (Yahoo Finance)."""
-    try:
-        # yfinance uses standardized symbols (e.g., BTC-USD)
-        ticker = yf.Ticker(symbol)
-        # Handle different intervals.  Yahoo Finance has limitations.
-        if interval == "1h":
-          period = "1d"  #  yfinance doesn't support 1h for historical data, so we'll use 1d and resample.
-          df = ticker.history(period=period, interval="1h")
-        elif interval == "1d":
-          df = ticker.history(period="1y", interval=interval) # Get 1 year of data
-        else:
-            raise ValueError("Invalid interval for yfinance.")
-        if df.empty:
-           raise Exception("No data returned from yfinance.")
-
-        df.reset_index(inplace=True)
-        df.rename(columns={"Datetime": "timestamp", "Open": "open", "High": "high", "Low": "low", "Close": "close", "Volume": "volume"}, inplace=True)
-        df = df[["timestamp", "open", "high", "low", "close", "volume"]]  # Select and order columns
-        return df.dropna()
-    except Exception as e:
-        raise Exception(f"Error fetching crypto data from yfinance: {e}")
+# Initialize pre-trained sentiment analysis model from Hugging Face Transformers
+sentiment_analyzer = pipeline("sentiment-analysis")
 
-def fetch_stock_data(symbol, interval="1d"):  # Simplified for yfinance
-    """Fetch stock market data using yfinance."""
+def fetch_news(keyword):
+    """Fetches news articles using the NewsAPI."""
     try:
-        ticker = yf.Ticker(symbol)
-        df = ticker.history(period="1y", interval=interval)  # Get 1 year of daily data
-
-        if df.empty:
-            raise Exception("No data returned from yfinance.")
-
-        df.reset_index(inplace=True)
-        df.rename(columns={"Date": "timestamp", "Open": "open", "High": "high", "Low": "low", "Close": "close", "Volume": "volume"}, inplace=True)
-        df = df[["timestamp", "open", "high", "low", "close", "volume"]]
-        return df.dropna()
+        url = f"https://newsapi.org/v2/everything?q={keyword}&apiKey={NEWS_API_KEY}"
+        response = requests.get(url)
+        response.raise_for_status()  # Raise an exception for HTTP errors
+        articles = response.json().get("articles", [])
+        return pd.DataFrame([{"title": article["title"], "description": article["description"]} for article in articles])
+    except requests.exceptions.RequestException as e:
+        raise Exception(f"Failed to fetch news: {e}")
     except Exception as e:
-        raise Exception(f"Error fetching stock data from yfinance: {e}")
+        raise Exception(f"Error processing news data: {e}")
 
-
-def fetch_sentiment_data(keyword):  # Placeholder - replace with a real sentiment analysis method
-    """Analyze sentiment from social media (placeholder)."""
+def fetch_social_media_data(keyword):
+    """Fetches social media data using Twitter API."""
     try:
-        tweets = [
-            f"{keyword} is going to moon!",
-            f"I hate {keyword}, it's trash!",
-            f"{keyword} is amazing!"
-        ]
-        sentiments = [TextBlob(tweet).sentiment.polarity for tweet in tweets]
-        return sum(sentiments) / len(sentiments) if sentiments else 0  # Avoid ZeroDivisionError
+        url = f"https://api.twitter.com/2/tweets/search/recent?query={keyword}&tweet.fields=text&max_results=10"
+        headers = {"Authorization": f"Bearer {TWITTER_BEARER_TOKEN}"}
+        response = requests.get(url, headers=headers)
+        response.raise_for_status()  # Raise an exception for HTTP errors
+        tweets = response.json().get("data", [])
+        if tweets: # Handle case when no tweets are found
+            return pd.DataFrame([{"text": tweet["text"]} for tweet in tweets])
+        else:
+            return pd.DataFrame({"text": []}) # Return empty DataFrame if no tweets found
+    except requests.exceptions.RequestException as e:
+        raise Exception(f"Failed to fetch social media data: {e}")
     except Exception as e:
-        print(f"Sentiment analysis error: {e}")
-        return 0
-
-# --- Technical Analysis Functions ---
-def calculate_technical_indicators(df):
-    """Calculates RSI, MACD, and Bollinger Bands."""
-    if df.empty:
-        return df
-
-    # RSI Calculation
-    delta = df['close'].diff()
-    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
-    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
-    rs = gain / loss
-    df['RSI'] = 100 - (100 / (1 + rs))
+        raise Exception(f"Error processing social media data: {e}")
 
-    # MACD Calculation
-    exp1 = df['close'].ewm(span=12, adjust=False).mean()
-    exp2 = df['close'].ewm(span=26, adjust=False).mean()
-    df['MACD'] = exp1 - exp2
-    df['Signal_Line'] = df['MACD'].ewm(span=9, adjust=False).mean()
-
-    # Bollinger Bands Calculation
-    df['MA20'] = df['close'].rolling(window=20).mean()
-    df['BB_upper'] = df['MA20'] + 2 * df['close'].rolling(window=20).std()
-    df['BB_lower'] = df['MA20'] - 2 * df['close'].rolling(window=20).std()
-
-    return df
-
-def create_technical_charts(df):
-    """Creates technical analysis charts (Price, RSI, MACD)."""
-    if df.empty:
-        return None, None, None
-
-    fig1 = go.Figure()
-    fig1.add_trace(go.Candlestick(
-        x=df['timestamp'],
-        open=df['open'],
-        high=df['high'],
-        low=df['low'],
-        close=df['close'],
-        name='Price'
-    ))
-    fig1.add_trace(go.Scatter(x=df['timestamp'], y=df['BB_upper'], name='Upper BB', line=dict(color='gray', dash='dash')))
-    fig1.add_trace(go.Scatter(x=df['timestamp'], y=df['BB_lower'], name='Lower BB', line=dict(color='gray', dash='dash')))
-    fig1.update_layout(title='Price and Bollinger Bands', xaxis_title='Date', yaxis_title='Price')
-
-    fig2 = go.Figure()
-    fig2.add_trace(go.Scatter(x=df['timestamp'], y=df['RSI'], name='RSI'))
-    fig2.add_hline(y=70, line_dash="dash", line_color="red")
-    fig2.add_hline(y=30, line_dash="dash", line_color="green")
-    fig2.update_layout(title='RSI Indicator', xaxis_title='Date', yaxis_title='RSI')
-
-    fig3 = go.Figure()
-    fig3.add_trace(go.Scatter(x=df['timestamp'], y=df['MACD'], name='MACD'))
-    fig3.add_trace(go.Scatter(x=df['timestamp'], y=df['Signal_Line'], name='Signal Line'))
-    fig3.update_layout(title='MACD', xaxis_title='Date', yaxis_title='Value')
-
-    return fig1, fig2, fig3
-
-# --- Prophet Forecasting Functions ---
-def prepare_data_for_prophet(df):
-    """Prepares data for Prophet."""
-    if df.empty:
-        return pd.DataFrame(columns=["ds", "y"])
-    df_prophet = df.rename(columns={"timestamp": "ds", "close": "y"})
-    return df_prophet[["ds", "y"]]
-
-def prophet_forecast(df_prophet, periods=10, freq="h", daily_seasonality=False, weekly_seasonality=False, yearly_seasonality=False, seasonality_mode="additive", changepoint_prior_scale=0.05):
-    """Performs Prophet forecasting."""
-    if df_prophet.empty:
-        return pd.DataFrame(), "No data for Prophet."
 
+def fetch_market_data(ticker, timeframe):
+    """Fetches stock/crypto market data using yfinance."""
     try:
-        model = Prophet(
-            daily_seasonality=daily_seasonality,
-            weekly_seasonality=weekly_seasonality,
-            yearly_seasonality=yearly_seasonality,
-            seasonality_mode=seasonality_mode,
-            changepoint_prior_scale=changepoint_prior_scale,
-        )
-        model.fit(df_prophet)
-        future = model.make_future_dataframe(periods=periods, freq=freq)
-        forecast = model.predict(future)
-        return forecast, ""
+        data = yf.download(ticker, period=timeframe, interval="1d")
+        if data.empty:
+            raise Exception(f"No market data found for ticker: {ticker}")
+        return data.reset_index()
     except Exception as e:
-        return pd.DataFrame(), f"Forecast error: {e}"
-
-def prophet_wrapper(df_prophet, forecast_steps, freq, daily_seasonality, weekly_seasonality, yearly_seasonality, seasonality_mode, changepoint_prior_scale):
-    """Wrapper for Prophet forecasting."""
-    if len(df_prophet) < 10:
-        return pd.DataFrame(), "Not enough data for forecasting (need >=10 rows)."
-
-    full_forecast, err = prophet_forecast(
-        df_prophet,
-        periods=forecast_steps,
-        freq=freq,
-        daily_seasonality=daily_seasonality,
-        weekly_seasonality=weekly_seasonality,
-        yearly_seasonality=yearly_seasonality,
-        seasonality_mode=seasonality_mode,
-        changepoint_prior_scale=changepoint_prior_scale,
-    )
-    if err:
-        return pd.DataFrame(), err
-
-    future_only = full_forecast.loc[len(df_prophet):, ["ds", "yhat", "yhat_lower", "yhat_upper"]]
-    return future_only, ""
-
-def create_forecast_plot(forecast_df):
-    """Creates the forecast plot."""
-    if forecast_df.empty:
-        return go.Figure()
-
-    fig = go.Figure()
-    fig.add_trace(go.Scatter(
-        x=forecast_df["ds"],
-        y=forecast_df["yhat"],
-        mode="lines",
-        name="Forecast",
-        line=dict(color="blue", width=2)
-    ))
-
-    fig.add_trace(go.Scatter(
-        x=forecast_df["ds"],
-        y=forecast_df["yhat_lower"],
-        fill=None,
-        mode="lines",
-        line=dict(width=0),
-        showlegend=True,
-        name="Lower Bound"
-    ))
-
-    fig.add_trace(go.Scatter(
-        x=forecast_df["ds"],
-        y=forecast_df["yhat_upper"],
-        fill="tonexty",
-        mode="lines",
-        line=dict(width=0),
-        name="Upper Bound"
-    ))
-
-    fig.update_layout(
-        title="Price Forecast",
-        xaxis_title="Time",
-        yaxis_title="Price",
-        hovermode="x unified",
-        template="plotly_white",
-    )
-    return fig
-
-# --- Model Training and Prediction ---
-model = RandomForestClassifier()  # Moved here
-
-def train_model(df):
-    """Train the AI model."""
-    if df.empty:
-        return  # Or raise an exception, or return a default model.
-    df["target"] = (df["close"].pct_change() > 0.05).astype(int)  # Target: 1 if price increased by >5%
-    features = df[["close", "volume"]].dropna()
-    target = df["target"].dropna()
-    if not features.empty and not target.empty: #check data is available
-      model.fit(features, target)
-    else:
-        print("Not enough data for model training.")
-
-def predict_growth(latest_data):
-    """Predict explosive growth."""
-    if not hasattr(model, 'estimators_') or len(model.estimators_) == 0:  # Check if model is trained
-        return [0]  # Or return an error message, or a default value
+        raise Exception(f"Failed to fetch market data for {ticker}: {str(e)}")
 
+def analyze_sentiment(text_list):
+    """Performs sentiment analysis on a list of texts."""
+    if not text_list:  # Handle empty text list
+        return []
     try:
-        prediction = model.predict(latest_data.reshape(1, -1))
-        return prediction
+        sentiments = sentiment_analyzer(text_list)
+        scores = [item['score'] if item['label'] == 'POSITIVE' else -item['score'] for item in sentiments]
+        return scores
     except Exception as e:
-        print(f"Prediction error: {e}")
-        return [0]
+        raise Exception(f"Sentiment analysis error: {e}")
 
-# --- Main Prediction and Display Function ---
-def analyze_market(market_type, symbol, interval, forecast_steps, daily_seasonality, weekly_seasonality, yearly_seasonality, seasonality_mode, changepoint_prior_scale, sentiment_keyword=""):
-    """Main function to orchestrate data fetching, analysis, and prediction."""
-    df = pd.DataFrame()
-    error_message = ""
-    sentiment_score = 0  # Initialize sentiment score
-    # 1. Data Fetching
-    try:
-        if market_type == "Crypto":
-            df = fetch_crypto_data(symbol, interval=interval)
-        elif market_type == "Stock":
-            df = fetch_stock_data(symbol, interval=interval)
-        else:
-            error_message = "Invalid market type selected."
-            return None, None, None, None, None, "", error_message, 0  # Also return sentiment
 
-        if sentiment_keyword: # If a keyword for sentiment is entered:
-            sentiment_score = fetch_sentiment_data(sentiment_keyword)
+def detect_anomalies(data):
+    """Detects anomalies in time series data using Isolation Forest."""
+    if len(data) <= 1: # Need at least 2 data points for diff and anomaly detection
+        return []
+    try:
+        model = IsolationForest(contamination=0.1, random_state=42)
+        anomalies = model.fit_predict(data.reshape(-1, 1))
+        return [i for i, val in enumerate(anomalies) if val == -1]
     except Exception as e:
-        error_message = f"Data Fetching Error: {e}"
-        return None, None, None, None, None, "", error_message, 0 #Return error + sentiment
-
-    if df.empty:
-        error_message = "No data fetched."
-        return None, None, None, None, None, "", error_message, 0  # Return empty + sentiment
-    # 2. Preprocessing & Technical Analysis
-    df["timestamp"] = pd.to_datetime(df["timestamp"])
-    numeric_cols = ["open", "high", "low", "close", "volume"]
-    df[numeric_cols] = df[numeric_cols].astype(float)
-    df = calculate_technical_indicators(df)
-
-    # 3. Prophet Forecasting
-    df_prophet = prepare_data_for_prophet(df)
-    freq = "h" if interval == "1h" or interval == "60min" else "d" #dynamic freq
-    forecast_df, prophet_error = prophet_wrapper(
-        df_prophet,
-        forecast_steps,
-        freq,
-        daily_seasonality,
-        weekly_seasonality,
-        yearly_seasonality,
-        seasonality_mode,
-        changepoint_prior_scale,
-    )
-
-    if prophet_error:
-        error_message = f"Prophet Error: {prophet_error}"
-        return None, None, None, None, None, "", error_message, sentiment_score #Return prophet error
+        raise Exception(f"Anomaly detection error: {e}")
+
+def identify_opportunities(ticker, news_sentiment, social_sentiment, anomalies, market_data):
+    """Identifies potential explosive growth opportunities."""
+    if np.mean(news_sentiment) > 0.3 and np.mean(social_sentiment) > 0.3 and len(anomalies) > 0: # Reduced sentiment threshold slightly
+        return [
+            {
+                "ticker": ticker,
+                "potential_gain": np.random.randint(10, 50),  # Simulated gain percentage
+                "risk_level": "High",
+                "disclaimer": "This is a speculative opportunity. Conduct thorough research."
+            }
+        ]
+    return []
 
-    forecast_plot = create_forecast_plot(forecast_df)
+def analyze_market(ticker_or_keyword, timeframe="1d"):
+    """
+    Analyzes news, social media, and market data for a given ticker or keyword.
 
-    # 4. Create the Charts
-    tech_plot, rsi_plot, macd_plot = create_technical_charts(df)
+    Args:
+        ticker_or_keyword (str): The stock ticker symbol (e.g., "AAPL") or keyword (e.g., "AI").
+        timeframe (str): The time frame for analysis (e.g., "1d", "1w", "1m").
 
-    # 5. Model Training and Prediction (simplified)
+    Returns:
+        dict: A dictionary containing analysis results for Gradio display.
+    """
     try:
-        train_model(df.copy()) # Train on a copy to avoid modifying original df.
-        if not df.empty: #Check if dataframe is empty or not.
-             latest_data = df[["close", "volume"]].iloc[-1].values # Get the last row for prediction.
-             growth_prediction = predict_growth(latest_data)
-             growth_label = "Yes" if growth_prediction[0] == 1 else "No"
-        else:
-            growth_label = "N/A: Insufficient Data" # If there is no data to predict the growth.
+        # Data Collection
+        news_df = fetch_news(ticker_or_keyword)
+        social_media_df = fetch_social_media_data(ticker_or_keyword)
+        market_df = fetch_market_data(ticker_or_keyword, timeframe)
+
+        # Sentiment Analysis
+        news_sentiment = analyze_sentiment(news_df["description"].fillna("").tolist()) if not news_df.empty else []
+        social_sentiment = analyze_sentiment(social_media_df["text"].tolist()) if not social_media_df.empty else []
+
+        # Anomaly Detection
+        price_changes = market_df["Close"].pct_change().dropna().values if not market_df.empty else np.array([])
+        anomalies = detect_anomalies(price_changes)
+
+        # Opportunity Identification
+        opportunities = identify_opportunities(
+            ticker_or_keyword,
+            news_sentiment,
+            social_sentiment,
+            anomalies,
+            market_df
+        )
 
-    except Exception as e:
-        error_message = f"Model Error: {e}"
-        growth_label = "N/A"
+        # Results Formatting for Gradio
+        results_md = f"## Analysis Results for: {ticker_or_keyword}\n\n"
 
-    # Prepare forecast data for the Dataframe output
-    forecast_df_display = forecast_df.loc[:, ["ds", "yhat", "yhat_lower", "yhat_upper"]].copy()
-    forecast_df_display.rename(columns={"ds": "Date", "yhat": "Forecast", "yhat_lower": "Lower Bound", "yhat_upper": "Upper Bound"}, inplace=True)
-    return forecast_plot, tech_plot, rsi_plot, macd_plot, forecast_df_display, growth_label, error_message, sentiment_score
+        results_md += f"**Average News Sentiment:** {np.mean(news_sentiment):.2f}  \n" if news_sentiment else "**Average News Sentiment:** N/A (No news found)  \n"
+        results_md += f"**Average Social Sentiment:** {np.mean(social_sentiment):.2f}  \n" if social_sentiment else "**Average Social Sentiment:** N/A (No social media data found)  \n"
+        results_md += f"**Anomalies Detected in Price Changes:** {len(anomalies)}  \n\n" if price_changes.size > 0 else "**Anomalies Detected in Price Changes:** N/A (Insufficient market data)  \n\n"
 
-# --- Gradio Interface ---
-with gr.Blocks(theme=gr.themes.Base()) as demo:
-    gr.Markdown("# Market Analysis and Prediction")
 
-    with gr.Row():
-        with gr.Column():
-            market_type_dd = gr.Radio(label="Market Type", choices=["Crypto", "Stock"], value="Crypto")
-            symbol_dd = gr.Dropdown(label="Symbol", choices=CRYPTO_SYMBOLS, value="BTC-USD")  # Use standardized symbols
-            interval_dd = gr.Dropdown(label="Interval", choices=INTERVAL_OPTIONS, value="1h")
-            forecast_steps_slider = gr.Slider(label="Forecast Steps", minimum=1, maximum=100, value=24, step=1)
-            daily_box = gr.Checkbox(label="Daily Seasonality", value=True)
-            weekly_box = gr.Checkbox(label="Weekly Seasonality", value=True)
-            yearly_box = gr.Checkbox(label="Yearly Seasonality", value=False)
-            seasonality_mode_dd = gr.Dropdown(label="Seasonality Mode", choices=["additive", "multiplicative"], value="additive")
-            changepoint_scale_slider = gr.Slider(label="Changepoint Prior Scale", minimum=0.01, maximum=1.0, step=0.01, value=0.05)
-            sentiment_keyword_txt = gr.Textbox(label="Sentiment Keyword (optional)") #Add Sentiment input
+        if opportunities:
+            results_md += "### Potential Explosive Growth Opportunities:\n"
+            opportunities_df = pd.DataFrame(opportunities)
+            results_md += opportunities_df.to_markdown(index=False) + "\n\n"
+        else:
+            results_md += "**No Explosive Growth Opportunities Identified based on current analysis.**\n\n"
 
-        with gr.Column():
-            forecast_plot = gr.Plot(label="Price Forecast")
-            with gr.Row():
-                tech_plot = gr.Plot(label="Technical Analysis")
-                rsi_plot = gr.Plot(label="RSI Indicator")
-            with gr.Row():
-                macd_plot = gr.Plot(label="MACD")
-            forecast_df = gr.Dataframe(label="Forecast Data", headers=["Date", "Forecast", "Lower Bound", "Upper Bound"])
-            growth_label_output = gr.Label(label="Explosive Growth Prediction") # Added for prediction.
-            sentiment_label_output = gr.Number(label="Sentiment Score") # Added for sentiment output
+        results_md += "---\n**Disclaimer:** This analysis is for informational purposes only and not financial advice. Investing in financial markets involves risk. Conduct thorough research and consult with a financial advisor before making investment decisions."
 
-    # Event Listener to update symbol dropdown based on market type
-    def update_symbol_choices(market_type):
-        if market_type == "Crypto":
-            return gr.Dropdown(choices=CRYPTO_SYMBOLS, value="BTC-USD")
-        elif market_type == "Stock":
-            return gr.Dropdown(choices=STOCK_SYMBOLS, value="AAPL") # Default to AAPL for stock
-        return gr.Dropdown(choices=[], value=None)  # Shouldn't happen, but safety check
-    market_type_dd.change(fn=update_symbol_choices, inputs=[market_type_dd], outputs=[symbol_dd])
+        return results_md
 
-    analyze_button = gr.Button("Analyze Market", variant="primary")
-    analyze_button.click(
-        fn=analyze_market,
-        inputs=[
-            market_type_dd,
-            symbol_dd,
-            interval_dd,
-            forecast_steps_slider,
-            daily_box,
-            weekly_box,
-            yearly_box,
-            seasonality_mode_dd,
-            changepoint_scale_slider,
-            sentiment_keyword_txt,  # Add sentiment keyword to the input
-        ],
-        outputs=[forecast_plot, tech_plot, rsi_plot, macd_plot, forecast_df, growth_label_output, gr.Label(label="Error Message"), sentiment_label_output] # Add sentiment score to the output
-    )
+    except Exception as e:
+        error_md = f"## Analysis Error for: {ticker_or_keyword}\n\n"
+        error_md += f"**Error Details:** {str(e)}\n\n"
+        error_md += "---\n**Disclaimer:** This analysis is for informational purposes only and not financial advice. Investing in financial markets involves risk. Conduct thorough research and consult with a financial advisor before making investment decisions."
+        return error_md
+
+
+# Gradio Interface
+iface = gr.Interface(
+    fn=analyze_market,
+    inputs=[
+        gr.Textbox(label="Stock Ticker or Keyword (e.g., AAPL, BTC-USD, AI)"),
+        gr.Dropdown(["1d", "1w", "1m"], label="Timeframe", value="1d"),
+    ],
+    outputs=gr.Markdown(label="Analysis Results"),
+    title="Explosive Growth Opportunity Finder",
+    description=(
+        "This tool leverages AI to analyze news sentiment, social media trends, and market data to identify potential investment opportunities. "
+        "Enter a stock ticker (e.g., AAPL), crypto symbol (e.g., BTC-USD), or a general keyword (e.g., AI) to analyze. "
+        "**Disclaimer:** This is a highly speculative tool for educational purposes.  "
+        "It is not financial advice. Investing in financial markets involves significant risk. "
+        "Always conduct your own thorough research and consult with a financial advisor before making any investment decisions."
+    ),
+)
 
 if __name__ == "__main__":
-    demo.launch()
+    iface.launch()