app.py

import os
import requests
from typing import Dict, Any
import snowflake.connector
from openai import OpenAI
import gradio as gr
import time
from datetime import datetime
from io import BytesIO
import base64
import matplotlib.pyplot as plt
import pandas as pd

def get_financial_data(ticker: str, api_key: str, period: str = 'annual') -> Dict[str, Any]:
    base_url = "https://financialmodelingprep.com/api/v3/"
    endpoints = {
        "profile": f"profile/{ticker}?apikey={api_key}",
        "balance_sheet": f"balance-sheet-statement/{ticker}?period={period}&apikey={api_key}",
        "income_statement": f"income-statement/{ticker}?period={period}&apikey={api_key}"
    }

    results = {}

    for key, endpoint in endpoints.items():
        url = base_url + endpoint
        response = requests.get(url)
        if response.status_code == 200:
            data = response.json()
            if data:
                if key == "profile":
                    results[key] = data[0]
                elif key == "balance_sheet":
                    results[key] = data[0]  # Get the most recent balance sheet statement
                elif key == "income_statement":
                    results[key] = data[0]  # Get the most recent income statement
            else:
                results[key] = None
        else:
            raise Exception(f"Failed to fetch data from {key} endpoint for ticker {ticker}. HTTP Status Code: {response.status_code}")

    return results

def get_proprietary_indicators(ticker: str) -> str:
    connection = snowflake.connector.connect(
        user=os.getenv('SNOWFLAKE_USER'),
        password=os.getenv('SNOWFLAKE_PASSWORD'),
        account=os.getenv('SNOWFLAKE_ACCOUNT'),
        warehouse=os.getenv('SNOWFLAKE_WAREHOUSE'),
        database=os.getenv('SNOWFLAKE_DATABASE')
    )
    query = f"""
    SELECT prompt
    FROM researchdata.TS_INDICATORS_FOR_GPT_PROMPT
    WHERE SYMBOL = '{ticker}'
    """
    cursor = connection.cursor()
    cursor.execute(query)
    result = cursor.fetchone()
    connection.close()
    
    if result:
        return result[0]
    else:
        return "No proprietary indicators available for this ticker."

def fetch_news(ticker: str, api_key: str) -> str:
    url = f'https://financialmodelingprep.com/api/v3/stock_news?page=0&tickers={ticker}&apikey={api_key}'
    response = requests.get(url)
    if response.status_code == 200:
        news_data = response.json()
        if news_data:
            news_summary = '\n\nRecent News Articles:\n'
            for article in news_data[:5]:  # Limit to the 5 most recent articles
                news_summary += f"Publish Date: {article['publishedDate']}\nTitle: {article['title']}\nSummary:{article['text']}\n\n"
            return news_summary
        else:
            return "No recent news articles found."
    else:
        return "Failed to fetch news articles."

def format_for_gpt_prompt(financial_data: Dict[str, Any], proprietary_indicators: str, news_data: str) -> str:
    profile = financial_data.get('profile', {})
    balance_sheet = financial_data.get('balance_sheet', {})
    income_statement = financial_data.get('income_statement', {})

    company_name = profile.get('companyName', 'N/A')
    company_description = profile.get('description', 'N/A')
    ticker = profile.get('symbol', 'N/A')
    market_cap = profile.get('mktCap', 'N/A')
    ceo = profile.get('ceo', 'N/A')
    beta = profile.get('beta', 'N/A')

    financial_data_str = f"""
Balance Sheet:
Total Assets: {balance_sheet.get('totalAssets', 'N/A')}
Total Liabilities: {balance_sheet.get('totalLiabilities', 'N/A')}
Total Equity: {balance_sheet.get('totalEquity', 'N/A')}
Cash and Cash Equivalents: {balance_sheet.get('cashAndCashEquivalents', 'N/A')}
Cash and Short Term Investments: {balance_sheet.get('cashAndShortTermInvestments', 'N/A')}
Inventory: {balance_sheet.get('inventory', 'N/A')}
Net Receivables: {balance_sheet.get('netReceivables', 'N/A')}
Total Current Assets: {balance_sheet.get('totalCurrentAssets', 'N/A')}
Total Current Liabilities: {balance_sheet.get('totalCurrentLiabilities', 'N/A')}
Long Term Debt: {balance_sheet.get('longTermDebt', 'N/A')}
Short Term Debt: {balance_sheet.get('shortTermDebt', 'N/A')}
Total Debt: {balance_sheet.get('totalDebt', 'N/A')}
Property, Plant, and Equipment Net: {balance_sheet.get('propertyPlantEquipmentNet', 'N/A')}
Deferred Revenue: {balance_sheet.get('deferredRevenue', 'N/A')}
Accumulated Other Comprehensive Income Loss: {balance_sheet.get('accumulatedOtherComprehensiveIncomeLoss', 'N/A')}
Retained Earnings: {balance_sheet.get('retainedEarnings', 'N/A')}
Total Non-Current Assets: {balance_sheet.get('totalNonCurrentAssets', 'N/A')}
Total Non-Current Liabilities: {balance_sheet.get('totalNonCurrentLiabilities', 'N/A')}
Goodwill and Intangible Assets: {balance_sheet.get('goodwillAndIntangibleAssets', 'N/A')}

Income Statement:
Revenue: {income_statement.get('revenue', 'N/A')}
Cost of Revenue: {income_statement.get('costOfRevenue', 'N/A')}
Gross Profit: {income_statement.get('grossProfit', 'N/A')}
Gross Profit Ratio: {income_statement.get('grossProfitRatio', 'N/A')}
Operating Income: {income_statement.get('operatingIncome', 'N/A')}
Operating Income Ratio: {income_statement.get('operatingIncomeRatio', 'N/A')}
Net Income: {income_statement.get('netIncome', 'N/A')}
Net Income Ratio: {income_statement.get('netIncomeRatio', 'N/A')}
EBITDA: {income_statement.get('ebitda', 'N/A')}
EBITDA Ratio: {income_statement.get('ebitdaratio', 'N/A')}
Earnings Per Share (EPS): {income_statement.get('eps', 'N/A')}
Diluted EPS: {income_statement.get('epsdiluted', 'N/A')}
Cost and Expenses: {income_statement.get('costAndExpenses', 'N/A')}
Depreciation and Amortization: {income_statement.get('depreciationAndAmortization', 'N/A')}
Income Before Tax: {income_statement.get('incomeBeforeTax', 'N/A')}
Income Before Tax Ratio: {income_statement.get('incomeBeforeTaxRatio', 'N/A')}
Income Tax Expense: {income_statement.get('incomeTaxExpense', 'N/A')}
Operating Expenses: {income_statement.get('operatingExpenses', 'N/A')}
Research and Development Expenses: {income_statement.get('researchAndDevelopmentExpenses', 'N/A')}
Selling, General, and Administrative Expenses: {income_statement.get('sellingGeneralAndAdministrativeExpenses', 'N/A')}
Total Other Income Expenses Net: {income_statement.get('totalOtherIncomeExpensesNet', 'N/A')}
Weighted Average Shares Outstanding: {income_statement.get('weightedAverageShsOut', 'N/A')}
Weighted Average Shares Outstanding Diluted: {income_statement.get('weightedAverageShsOutDil', 'N/A')}
"""

    prompt = f"""
Here is the profile of the company you will be analyzing:

<company>
Name: {company_name} ({ticker})
Description: {company_description}
Market Cap: {market_cap}
CEO: {ceo}
Beta: {beta}
</company>

Please carefully review this background information on the company.

Next, here is the most recent quarter's balance sheet and income statement data for the company:

<financial_data>
{financial_data_str}
</financial_data>

Next, here are some of the proprietary TradeSmith indicators for the company:

<proprietary_indicators>

{proprietary_indicators}

</proprietary_indicators>

Next, here are some recent news articles about the company:

<news_articles>

{news_data}

</news_articles>

<analysis>

Provide your analysis of the company profile, financial data, proprietary TradeSmith indicators, and recent news. Show your analytical thought process.

</analysis>

<prediction>

Considering everything, what prediction would you make for this stock's return over the next 21 trading days? If TradeSmith's Predictive Alpha projection was told to you, use that as your prediction, and make sure your reasoning aligns with the prediction. Otherwise, make your own prediction based on the information you analyzed.

</prediction>

Remember, your role is to be an expert stock analyst. Explain your reasoning so that investors can understand the key factors driving your recommendation. Assume you are talking to a group of readers, not to an individual, to comply with legal's requirement of not giving out personalized advice. At the end, always disclaim something along the lines of "investing and trading is risky..."
"""

    return prompt

def generate_gpt_prompt(ticker: str, api_key: str) -> str:
    financial_data = get_financial_data(ticker, api_key)
    proprietary_indicators = get_proprietary_indicators(ticker)
    news_data = fetch_news(ticker, api_key)
    return format_for_gpt_prompt(financial_data, proprietary_indicators, news_data)

def run_gpt_model(prompt: str, api_key: str) -> str:
    openai_client = OpenAI(api_key = api_key)

    system_prompt = """
You are a highly experienced stock analyst with deep knowledge of TradeSmith and InvestorPlace products and services. You assist subscribers of these platforms by providing comprehensive stock analyses. Ensure your output is clear, concise, and formatted for easy reading with line breaks after each sentence. The analysis should be structured as follows:

[Company Description]:
Provide a brief overview of the company in 2-3 sentences.

[TradeSmith Indicator Analysis]:
Incorporate relevant TradeSmith indicators, explaining what they suggest about the stock's performance and potential. Use proprietary indicators effectively to add value to the analysis. Use bullet points, and for each point, explain why it's worth noting. If any billionaires hold the stock, name them. If the Predictive Alpha prediction is told to you, don't talk about it here, use it as your prediction in the following Prediction & Analysis section.

[News Analysis]:
Summarize recent news articles or reports related to the company. Discuss any significant events or developments that could impact the stock's performance. Use bullet points, and for each point, explain why it's worth noting.

[Income Statement Analysis]:
Analyze key figures from the company's income statement, such as revenue, expenses, and net income. Highlight trends and significant changes. Use bullet points, and for each point, explain why it's worth noting.

[Balance Sheet Analysis]:
Examine important elements from the balance sheet, focusing on assets, liabilities, and shareholders' equity. Discuss the company's financial health and stability. Use bullet points, and for each point, explain why it's worth noting.

[Prediction & Reasoning]:
Provide a well-reasoned prediction for the stock's future 21 day performance. If a Predictive Alpha prediction was provided, use that as your prediction, supported by your analysis. Offer actionable recommendations based on your insights.

Make sure each section is thorough yet concise, providing valuable insights for the subscriber. Only touch on the details you deem important. Your output should look professionally formatted and work well with gradio's markup output.  

Finally, it is critically important that you comply with the legal guideline of not providing personalized financial advice to individuals.
"""


    system = {
      "role":"system",
      "content": (system_prompt)
      }

    user = {
      "role":"user",
      "content":prompt
      }

    response = openai_client.chat.completions.create(
        model="gpt-4o",  # or another model
        messages = [system,user],
        max_tokens=1500,
        temperature=0.7,
        stream=True
    )

    collected_chunks = ""
    for chunk in response:
        chunk_content = chunk.choices[0].delta.content or ""
        if chunk_content:
            yield chunk_content
            #time.sleep(0.01)

def plot_price_chart(ticker: str):
    # Establishing connection to Snowflake
    connection = snowflake.connector.connect(
        user=os.getenv('SNOWFLAKE_USER'),
        password=os.getenv('SNOWFLAKE_PASSWORD'),
        account=os.getenv('SNOWFLAKE_ACCOUNT'),
        warehouse=os.getenv('SNOWFLAKE_WAREHOUSE'),
        database=os.getenv('SNOWFLAKE_DATABASE')
    )

    cursor = connection.cursor()

    # Function to get historical price data
    def get_historical_price_data(ticker):
        query = f"""
        select s.symbol,
               sd.TRADEDATE,
               sd.TRADEADJCLOSE
            from HISTORICALDATANEW.SYMBOL_SVIEW s
        join HISTORICALDATANEW.SYMBOLTYPES_SVIEW st on s.SYMBOLTYPEID = st.SYMBOLTYPEID
        join HISTORICALDATANEW.STOCKDATA_SVIEW sd on s.SYMBOLID = sd.SYMBOLID
        where st.name in ('Common stock','ETF')
        and case when s.ISDUPLICATE then 1 else 0 end = 0
        and s.EXCHANGEID in (1,8)
        and s.SYMBOL = '{ticker}'
        qualify row_number() over (partition by s.SYMBOLID order by TRADEDATE desc) <= 50
        order by TRADEDATE;
        """
        historical_price_data = cursor.execute(query).fetch_pandas_all()
        return historical_price_data

    # Function to get predicted return data
    def get_predicted_return_data(ticker):
        query = f"""
        select s.SYMBOL,
               pa.PREDICTEDTRADEDATE,
               pa.PREDICTEDRETURN
        from RESEARCHDATA.PREDICTIVEALPHAHISTORICALVALUES_SVIEW pa
        join HISTORICALDATANEW.SYMBOL_SVIEW s on s.SYMBOLID = pa.SYMBOLID
        where pa.PREDICTEDPERIOD = 21
        and s.SYMBOL = '{ticker}'
        qualify row_number() over (partition by pa.SYMBOLID order by TRADEDATE desc) = 1;
        """
        predicted_return_data = cursor.execute(query).fetch_pandas_all()
        return predicted_return_data

    historical_data = get_historical_price_data(ticker)
    predicted_data = get_predicted_return_data(ticker)
    
    # Extract the relevant data
    historical_dates = pd.to_datetime(historical_data['TRADEDATE'])
    historical_prices = historical_data['TRADEADJCLOSE'].astype(float)
    
    last_known_date = historical_dates.iloc[-1]
    last_known_price = historical_prices.iloc[-1]
    
    predicted_date = pd.to_datetime(predicted_data['PREDICTEDTRADEDATE'].iloc[0])
    predicted_return = float(predicted_data['PREDICTEDRETURN'].iloc[0])
    
    predicted_price = last_known_price * (1 + predicted_return)
    
    # Plot the data
    plt.figure(figsize=(12, 6))  # Increase figure size to give more space for annotations
    plt.plot(historical_dates, historical_prices, label='Historical Prices', marker='o')
    plt.axvline(x=last_known_date, color='gray', linestyle='--', label='Last Known Date')
    plt.plot([last_known_date, predicted_date], [last_known_price, predicted_price], 'r--', label='Predicted Price')
    
    # Adding gap for 21 trading days (assuming 5 trading days per week)
    gap_dates = pd.date_range(start=last_known_date, periods=22, freq='B')[-1]
    plt.axvline(x=gap_dates, color='blue', linestyle='--', label='21 Trading Days Gap')

    # Annotations
    plt.annotate(f'Last Known Price: {last_known_price:.2f}', 
                 xy=(last_known_date, last_known_price), 
                 xytext=(last_known_date, last_known_price * 1.05),
                 arrowprops=dict(facecolor='black', arrowstyle='->'),
                 bbox=dict(boxstyle='round,pad=0.5', edgecolor='black', facecolor='white'))
    
    plt.annotate(f'Predicted Price: {predicted_price:.2f}\nPredicted Return: {predicted_return:.2%}', 
                 xy=(predicted_date, predicted_price), 
                 xytext=(predicted_date, predicted_price * 1.05),
                 arrowprops=dict(facecolor='red', arrowstyle='->'),
                 bbox=dict(boxstyle='round,pad=0.5', edgecolor='red', facecolor='white'))
    
    plt.xlabel('Date')
    plt.ylabel('Price')
    plt.title(f'{ticker} Price Chart with Predicted Price')
    plt.legend()
    plt.grid(True)
    
    # Adjust limits to ensure annotations fit
    plt.ylim(bottom=min(historical_prices) * 0.95, top=max(historical_prices) * 1.1)
    plt.xlim(left=min(historical_dates), right=predicted_date + pd.Timedelta(days=10))

    # Save plot to a bytes buffer
    buf = BytesIO()
    plt.savefig(buf, format="png")
    buf.seek(0)
    img_str = base64.b64encode(buf.read()).decode("utf-8")
    buf.close()
    
    return f'<img src="data:image/png;base64,{img_str}" alt="Price Chart">'

def analyze_stock(ticker: str):
    ticker = ticker.upper()
    fmp_api_key = os.getenv('FMP_API_KEY')
    openai_api_key = os.getenv('OPENAI_API_KEY')
    prompt = generate_gpt_prompt(ticker, fmp_api_key)
    yield prompt, "", ""  # Initially yield the prompt with empty analysis and chart response
    
    analysis_chunks = []
    for analysis_chunk in run_gpt_model(prompt, openai_api_key):
        analysis_chunks.append(analysis_chunk)
        yield "".join(analysis_chunks), prompt, ""  # Stream the accumulated analysis chunks
    
    # Generate the price chart after the analysis
    price_chart_html = plot_price_chart(ticker)
    yield "".join(analysis_chunks),price_chart_html, prompt

description = """We gave Ann-E a name, but not a personality. It's time to change that!

               Imagine entering a ticker on the TradeSmith site, going to the Predictive Alpha widget, and getting a stock analysis from Ann-E to go along with the prediction.
               
               Currently, the analysis is based on company profile information, recent news, the most recent quarter's balance sheet and income statements, and TradeSmith proprietary indicators.
               
               This demonstration shows just a simple use case for building out a proprietary Alta Large Language Model (LLM). 
               
               Future vision includes interactive model, fine-tuned on Alta content and indicators, that help our subscribers and keep them engaged on the site. 
               
               The possibilities are endless!
               
               Note: We can control outputs to meet legal requirements.
               """
              

# Gradio Interface
iface = gr.Interface(
    fn=analyze_stock, 
    inputs=gr.Textbox(label='Enter a Ticker', value='AAPL'), 
    outputs=[
        gr.Markdown(label='Ann-E Response'),
        gr.HTML(label='Price Chart'),
        gr.Textbox(label='Prompt for Model (behind the scenes).')
        
    ], 
    title="Demo: Predictive Alpha + Stock Analysis GPT Model",
    description=description
)

if __name__ == "__main__":
    iface.launch()