app.py

import gradio as gr
import requests
import os
from dotenv import load_dotenv
from groq import Groq
import pandas as pd
from datetime import datetime, timedelta


# API Keys from .env file
ALPHA_VANTAGE_API_KEY = os.environ.get("ALPHA_VANTAGE_API_KEY")
GROQ_API_KEY = os.environ.get("GROQ_API_KEY")

# Initialize Groq client
groq_client = Groq(api_key=GROQ_API_KEY)

# Get RSI data from Alpha Vantage
def get_rsi_data(stock_symbol):
    url = f"https://www.alphavantage.co/query?function=RSI&symbol={stock_symbol}&interval=daily&time_period=14&series_type=close&apikey={ALPHA_VANTAGE_API_KEY}"
    
    response = requests.get(url)
    data = response.json()
    
    try:
        if "Note" in data:
            return f"API Limit Reached: {data['Note']}", None
            
        latest_date = list(data["Technical Analysis: RSI"].keys())[0]
        rsi = float(data["Technical Analysis: RSI"][latest_date]["RSI"])
        
        result = f"Symbol: {stock_symbol}\n"
        result += f"Date: {latest_date}\n"
        result += f"RSI: {rsi:.2f}\n"
        
        if rsi < 30:
            initial_recommendation = "Potentially Oversold"
        elif rsi > 70:
            initial_recommendation = "Potentially Overbought"
        else:
            initial_recommendation = "Neutral RSI"
            
        result += f"Technical Indicator: {initial_recommendation}\n"
        return result, rsi
        
    except Exception as e:
        return f"Error fetching RSI data: {data.get('Note', str(e))}", None

# Get time series data from Alpha Vantage
def get_time_series_data(stock_symbol):
    url = f"https://www.alphavantage.co/query?function=TIME_SERIES_DAILY&symbol={stock_symbol}&outputsize=compact&apikey={ALPHA_VANTAGE_API_KEY}"
    
    response = requests.get(url)
    data = response.json()
    
    try:
        if "Note" in data:
            return f"API Limit Reached: {data['Note']}"
            
        if "Error Message" in data:
            return f"Error: {data['Error Message']}"
            
        time_series = data.get("Time Series (Daily)", {})
        if not time_series:
            return "Error: No time series data available"
            
        # Get recent data (last 10 trading days)
        dates = list(time_series.keys())[:10]
        
        # Format time series data
        result = f"\nRecent Price History for {stock_symbol}:\n"
        
        # Calculate some basic metrics
        closing_prices = []
        
        for date in dates:
            close_price = float(time_series[date]['4. close'])
            closing_prices.append(close_price)
            
            # Only add the last 5 days to keep the text shorter
            if len(closing_prices) <= 5:
                result += f"{date}: Close ${close_price:.2f}\n"
        
        # Calculate some trends
        if len(closing_prices) >= 2:
            latest_price = closing_prices[0]
            prev_price = closing_prices[1]
            day_change = (latest_price - prev_price) / prev_price * 100
            
            if len(closing_prices) >= 5:
                five_day_change = (latest_price - closing_prices[4]) / closing_prices[4] * 100
                result += f"\n5-Day Change: {five_day_change:.2f}%\n"
                
            result += f"1-Day Change: {day_change:.2f}%\n"
            
        return result
        
    except Exception as e:
        return f"Error fetching time series data: {str(e)}"

# Get LLM recommendation from Groq
def get_llm_recommendation(stock_symbol, rsi_data, time_series_data):
    if "Error" in rsi_data or "API Limit" in rsi_data:
        return rsi_data
    
    prompt = f"""
    As a financial advisor, analyze the following data for {stock_symbol}:
    
    RSI TECHNICAL DATA:
    {rsi_data}
    
    PRICE HISTORY:
    {time_series_data}
    
    Based on both the RSI technical indicator and the recent price history, provide a comprehensive recommendation (buy, sell, or hold) with a short explanation.
    Consider price trends, momentum, and RSI values in your analysis.
    Keep your response concise (3-5 sentences maximum).
    """
    
    try:
        # Using the Groq client instead of direct API calls
        completion = groq_client.chat.completions.create(
            model="llama-3.3-70b-versatile",
            messages=[
                {"role": "user", "content": prompt}
            ],
            temperature=0.7,
            max_tokens=150
        )
        
        # Extract the recommendation
        recommendation = completion.choices[0].message.content.strip()
        return recommendation
            
    except Exception as e:
        return f"Error from LLM service: {str(e)}"

# Main function for Gradio
def get_stock_recommendation(stock_symbol):
    # Get RSI data
    rsi_data, rsi_value = get_rsi_data(stock_symbol)
    
    # If there was an error, just return the error
    if rsi_value is None:
        return rsi_data, "Could not analyze without valid RSI data."
    
    # Get time series data
    time_series_data = get_time_series_data(stock_symbol)
    
    # Combine the data for display
    combined_data = rsi_data + "\n" + time_series_data
        
    # Get LLM recommendation
    llm_recommendation = get_llm_recommendation(stock_symbol, rsi_data, time_series_data)
    
    return combined_data, llm_recommendation

# Gradio UI
with gr.Blocks() as demo:
    gr.Markdown("## 📊 Enhanced Stock Analyzer")
    gr.Markdown("Uses RSI + Time Series data with LLaMA AI analysis")
    
    with gr.Row():
        stock_input = gr.Textbox(label="Enter Stock Symbol (e.g., AAPL, MSFT, GOOG)")
        submit_button = gr.Button("Analyze", variant="primary")
        
    with gr.Row():
        with gr.Column():
            technical_output = gr.Textbox(label="Technical Data", lines=12)
        with gr.Column():
            llm_output = gr.Textbox(label="LLaMA AI Recommendation", lines=12)
    
    gr.Markdown("*Note: Using Alpha Vantage (technical data) and Groq's LLaMA (AI recommendations)*")
    
    submit_button.click(
        fn=get_stock_recommendation, 
        inputs=stock_input, 
        outputs=[technical_output, llm_output]
    )

if __name__ == "__main__":
    if not ALPHA_VANTAGE_API_KEY or not GROQ_API_KEY:
        print("Error: Please ensure your API keys are set in the .env file")
        print("ALPHA_VANTAGE_API_KEY and GROQ_API_KEY are required")
    else:
        demo.launch()