app.py

import os
import time
from bs4 import BeautifulSoup
import re
import requests
import base64
import json
import yfinance as yf
import langchain
from langchain.agents import Tool, initialize_agent
from langchain.llms import HuggingFacePipeline
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
import streamlit as st
import warnings
warnings.filterwarnings("ignore")

# Function to load background image
def get_base64(bin_file):
    with open(bin_file, 'rb') as f:
        data = f.read()
    return base64.b64encode(data).decode()

def set_background(png_file):
    bin_str = get_base64(png_file)
    page_bg_img = '''
    <style>
    .stApp {
    background-image: url("data:image/png;base64,%s");
    background-size: cover;
    }
    </style>
    ''' % bin_str
    st.markdown(page_bg_img, unsafe_allow_html=True)


st.header('Stock Recommendation System')

# Sidebar for API key and instructions
st.sidebar.write('This tool provides recommendation based on the RAG & ReAct Based Schemes:')
lst = ['Get Ticker Value', 'Fetch Historic Data on Stock', 'Get Financial Statements', 
       'Scrape the Web for Stock News', 'LLM ReAct based Verbal Analysis', 
       'Output Recommendation: Buy, Sell, or Hold with Justification']
s = '\n'.join(f"- {item}" for item in lst)
st.sidebar.markdown(s)

# Load Mistral Model and Tokenizer
import os
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
from langchain import HuggingFacePipeline
import streamlit as st

@st.cache_resource
def load_model():
    token = os.getenv("token")# Set this environment variable securely
    model_name = "mistralai/Ministral-8B-Instruct-2410"
    tokenizer = AutoTokenizer.from_pretrained(model_name, use_auth_token=token)
    model = AutoModelForCausalLM.from_pretrained(model_name, use_auth_token=token)
    text_gen_pipeline = pipeline("text-generation", model=model, tokenizer=tokenizer)
    return HuggingFacePipeline(pipeline=text_gen_pipeline)

llm = load_model()

# Get Historical Stock Closing Price for Last 1 Year
def get_stock_price(ticker):
    if "." in ticker:
        ticker = ticker.split(".")[0]
    stock = yf.Ticker(ticker)
    df = stock.history(period="1y")
    df = df[["Close", "Volume"]]
    df.index = [str(x).split()[0] for x in list(df.index)]
    df.index.rename("Date", inplace=True)
    return df.to_string()

# Web Scraping for Recent News
def google_query(search_term):
    if "news" not in search_term:
        search_term += " stock news"
    url = f"https://www.google.com/search?q={search_term}"
    url = re.sub(r"\s", "+", url)
    return url

def get_recent_stock_news(company_name):
    headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.0.0 Safari/537.36'}
    g_query = google_query(company_name)
    res = requests.get(g_query, headers=headers).text
    soup = BeautifulSoup(res, "html.parser")
    news = [n.text for n in soup.find_all("div", "n0jPhd ynAwRc tNxQIb nDgy9d")]
    news += [n.text for n in soup.find_all("div", "IJl0Z")]

    if len(news) > 6:
        news = news[:4]
    
    news_string = "\n".join(f"{i}. {n}" for i, n in enumerate(news))
    return "Recent News:\n\n" + news_string

# Get Financial Statements
def get_financial_statements(ticker):
    if "." in ticker:
        ticker = ticker.split(".")[0]
    company = yf.Ticker(ticker)
    balance_sheet = company.balance_sheet
    if balance_sheet.shape[1] > 3:
        balance_sheet = balance_sheet.iloc[:, :3]
    balance_sheet = balance_sheet.dropna(how="any")
    return balance_sheet.to_string()

# Initialize DuckDuckGo Search Engine
from langchain_community.tools import DuckDuckGoSearchRun
search = DuckDuckGoSearchRun()

tools = [
    Tool(
        name="Stock Ticker Search",
        func=search.run,
        description="Use only when you need to get stock ticker from internet, you can also get recent stock related news. Don't use it for any other analysis or task."
    ),
    Tool(
        name="Get Stock Historical Price",
        func=get_stock_price,
        description="Use when you are asked to evaluate or analyze a stock. This will output historic share price data. You should input the stock ticker to it."
    ),
    Tool(
        name="Get Recent News",
        func=get_recent_stock_news,
        description="Use this to fetch recent news about stocks."
    ),
    Tool(
        name="Get Financial Statements",
        func=get_financial_statements,
        description="Use this to get financial statement of the company. With the help of this data company's historic performance can be evaluated. You should input stock ticker to it."
    )
]

zero_shot_agent = initialize_agent(
    llm=llm,
    agent="zero-shot-react-description",
    tools=tools,
    verbose=True,
    max_iteration=4,
    return_intermediate_steps=False
)

stock_prompt = """You are a financial advisor. Give stock recommendations for given query.
...
Question: {input}
Thought:{agent_scratchpad}"""

zero_shot_agent.agent.llm_chain.prompt.template = stock_prompt

# Handling user input in Streamlit
if prompt := st.chat_input():
    st.chat_message("user").write(prompt)
    with st.chat_message("assistant"):
        response = zero_shot_agent(f'Is {prompt} a good investment choice right now?')
        st.write(response["output"])