app.py

import gradio as gr
import plotly.graph_objects as go
import yfinance as yf
import pandas as pd
import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, LSTM

# Define periods
periods = {
    "1 Month": "1mo",
    "3 Months": "3mo",
    "6 Months": "6mo",
    "1 Year": "1y",
    "5 Years": "5y",
    "10 Years": "10y",
    "Max": "max"
}

def fetch_data(ticker, period):
    data = yf.download(ticker, period=periods[period])
    return data

def plot_technical_analysis(ticker, period, analysis_type, ma_length, candle_period):
    data = fetch_data(ticker, period)
    
    fig = go.Figure()
    
    if analysis_type == "Candlestick":
        fig.add_trace(go.Candlestick(x=data.index,
                                     open=data['Open'],
                                     high=data['High'],
                                     low=data['Low'],
                                     close=data['Close'],
                                     name='Candlestick',
                                     xperiod=candle_period))
    
    elif analysis_type == "Moving Average":
        data[f'MA{ma_length}'] = data['Close'].rolling(window=ma_length).mean()
        fig.add_trace(go.Scatter(x=data.index, y=data['Close'], mode='lines', name='Close Price'))
        fig.add_trace(go.Scatter(x=data.index, y=data[f'MA{ma_length}'], mode='lines', name=f'{ma_length}-day MA'))
    
    elif analysis_type == "Bollinger Bands":
        data['MA20'] = data['Close'].rolling(window=20).mean()
        data['stddev'] = data['Close'].rolling(window=20).std()
        data['upper'] = data['MA20'] + (data['stddev'] * 2)
        data['lower'] = data['MA20'] - (data['stddev'] * 2)
        fig.add_trace(go.Scatter(x=data.index, y=data['Close'], mode='lines', name='Close Price'))
        fig.add_trace(go.Scatter(x=data.index, y=data['upper'], mode='lines', name='Upper Band'))
        fig.add_trace(go.Scatter(x=data.index, y=data['lower'], mode='lines', name='Lower Band'))

    elif analysis_type == "RSI":
        delta = data['Close'].diff(1)
        gain = delta.where(delta > 0, 0)
        loss = -delta.where(delta < 0, 0)
    
        avg_gain = gain.rolling(window=14).mean()
        avg_loss = loss.rolling(window=14).mean()
    
        rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))
    
        fig.add_trace(go.Scatter(x=data.index, y=rsi, mode='lines', name='RSI'))
    
    elif analysis_type == "MACD":
        exp1 = data['Close'].ewm(span=12, adjust=False).mean()
        exp2 = data['Close'].ewm(span=26, adjust=False).mean()
        macd = exp1 - exp2
        signal = macd.ewm(span=9, adjust=False).mean()
    
        fig.add_trace(go.Scatter(x=data.index, y=macd, mode='lines', name='MACD'))
        fig.add_trace(go.Scatter(x=data.index, y=signal, mode='lines', name='Signal Line'))

    fig.update_layout(title=f"{analysis_type} Analysis for {ticker}", xaxis_title="Date", yaxis_title="Price", xaxis_rangeslider_visible=False)
    return fig

def plot_fundamental_analysis(ticker, analysis_type):
    stock = yf.Ticker(ticker)
    
    if analysis_type == "Financials":
        data = stock.financials
    elif analysis_type == "Balance Sheet":
        data = stock.balance_sheet
    elif analysis_type == "Cash Flow":
        data = stock.cashflow
    
    return data.to_html()

def train_predictive_model(ticker, period, epochs, batch_size, future_days):
    data = fetch_data(ticker, period)
    data['Close'] = data['Close'].fillna(method='ffill')
    
    close_prices = data['Close'].values
    close_prices = close_prices.reshape(-1, 1)
    
    # Normalize the data
    from sklearn.preprocessing import MinMaxScaler
    scaler = MinMaxScaler()
    close_prices = scaler.fit_transform(close_prices)
    
    # Prepare the data for LSTM
    X = []
    y = []
    time_step = 10
    for i in range(time_step, len(close_prices) - future_days):
        X.append(close_prices[i-time_step:i])
        y.append(close_prices[i + future_days])
    
    X = np.array(X)
    y = np.array(y)
    
    # Split the data
    split = int(0.8 * len(X))
    X_train, X_test = X[:split], X[split:]
    y_train, y_test = y[:split], y[split:]
    
    # Build the LSTM model
    model = Sequential()
    model.add(LSTM(units=50, return_sequences=True, input_shape=(time_step, 1)))
    model.add(LSTM(units=50, return_sequences=False))
    model.add(Dense(units=25))
    model.add(Dense(units=1))
    
    model.compile(optimizer='adam', loss='mean_squared_error')
    
    # Train the model
    model.fit(X_train, y_train, epochs=epochs, batch_size=batch_size)
    
    # Predict
    predictions = model.predict(X_test)
    predictions = scaler.inverse_transform(predictions)
    y_test = scaler.inverse_transform(y_test)
    
    # Forecast future values
    future_predictions = []
    last_data = close_prices[-time_step:]
    for _ in range(future_days):
        pred = model.predict(last_data.reshape(1, time_step, 1))
        future_predictions.append(pred[0, 0])
        last_data = np.append(last_data[1:], pred[0])
    
    future_predictions = scaler.inverse_transform(np.array(future_predictions).reshape(-1, 1))
    
    # Plot the results
    fig = go.Figure()
    fig.add_trace(go.Scatter(x=data.index[-len(y_test):], y=y_test.flatten(), mode='lines', name='Actual'))
    fig.add_trace(go.Scatter(x=data.index[-len(predictions):], y=predictions.flatten(), mode='lines', name='Predicted'))
    
    future_dates = pd.date_range(start=data.index[-1], periods=future_days + 1, inclusive='right')
    
    fig.add_trace(go.Scatter(x=future_dates, y=future_predictions.flatten(), mode='lines', name='Future Predictions'))
    
    fig.update_layout(title=f"Predicted vs Actual and Future Forecast for {ticker}", xaxis_title="Date", yaxis_title="Price", xaxis_rangeslider_visible=False)
    return fig

# Gradio Interface
with gr.Blocks() as demo:
    gr.Markdown("# Ultimate Asset Analyzer (Use Yahoo Finance Tickers)")
    
    with gr.Tab("Technical Analysis"):
        with gr.Row():
            ticker = gr.Textbox(label="Ticker Symbol", value="AAPL")
            period = gr.Dropdown(label="Period", choices=list(periods.keys()), value="1 Year")
        analysis_type = gr.Radio(label="Analysis Type", choices=["Candlestick", "Moving Average", "Bollinger Bands", "RSI", "MACD"], value="Candlestick")
        
        ma_length = gr.Number(label="Moving Average Length", value=20, visible=False)
        candle_period = gr.Number(label="Candlestick Period", value=1, visible=False)
        
        def update_visibility(analysis_type):
            return (
                gr.update(visible=analysis_type == "Moving Average"),
                gr.update(visible=analysis_type == "Candlestick")
            )
        
        analysis_type.change(fn=update_visibility, inputs=analysis_type, outputs=[ma_length, candle_period])
        
        plot_button = gr.Button("Plot")
        plot_output = gr.Plot()
        
        plot_button.click(fn=plot_technical_analysis, inputs=[ticker, period, analysis_type, ma_length, candle_period], outputs=plot_output)
    
    with gr.Tab("Fundamental Analysis"):
        ticker = gr.Textbox(label="Ticker Symbol", value="AAPL")
        analysis_type = gr.Radio(label="Analysis Type", choices=["Financials", "Balance Sheet", "Cash Flow"], value="Financials")
        plot_button = gr.Button("Show Data")
        table_output = gr.HTML()
        
        plot_button.click(fn=plot_fundamental_analysis, inputs=[ticker, analysis_type], outputs=table_output)
    
    with gr.Tab("Predictive Model"):
        ticker = gr.Textbox(label="Ticker Symbol", value="AAPL")
        period = gr.Dropdown(label="Period", choices=list(periods.keys()), value="1 Year")
        epochs = gr.Number(label="Epochs", value=10)
        batch_size = gr.Number(label="Batch Size", value=32)
        future_days = gr.Number(label="Days to Predict", value=30)
        train_button = gr.Button("Train and Predict")
        predict_output = gr.Plot()
        
        train_button.click(fn=train_predictive_model, inputs=[ticker, period, epochs, batch_size, future_days], outputs=predict_output)

# Launch the interface
demo.launch()