import os import nltk import requests # Use a directory within the user's home directory nltk_data_dir = os.path.expanduser("~/.nltk_data") os.makedirs(nltk_data_dir, exist_ok=True) nltk.data.path.append(nltk_data_dir) # Download NLTK data nltk.download('punkt', download_dir=nltk_data_dir, quiet=True) import chainlit as cl from llama_index.core import VectorStoreIndex, Document from llama_index.embeddings.huggingface import HuggingFaceEmbedding from llama_index.llms.groq import Groq from llama_index.core import ServiceContext from llama_index.core.node_parser import SentenceSplitter from dotenv import load_dotenv import yfinance as yf import pandas as pd load_dotenv() # Fetch the API keys from environment variables GROQ_API_KEY = os.getenv("GROQ_API_KEY") FMP_API_KEY = os.getenv("FMP_API_KEY") # Initialize models embed_model = HuggingFaceEmbedding(model_name="sentence-transformers/all-MiniLM-L6-v2") llm = Groq(model="llama3-70b-8192", api_key=GROQ_API_KEY) # Create service context service_context = ServiceContext.from_defaults( llm=llm, embed_model=embed_model, node_parser=SentenceSplitter(chunk_size=1000, chunk_overlap=200) ) def fetch_earnings_transcript(symbol: str) -> str: """ Fetch the latest transcript for a company's earnings call. Args: - symbol (str): The stock ticker symbol (e.g., 'AAPL'). Returns: - str: The earnings call transcript or an error message. """ transcript_url = f"https://financialmodelingprep.com/api/v3/earning_call_transcript/{symbol}?apikey={FMP_API_KEY}" try: response = requests.get(transcript_url, timeout=10) response.raise_for_status() transcript_data = response.json() if not transcript_data: return f"No transcript available for {symbol}." # Extract the first available transcript latest_transcript = transcript_data[0].get("content", "") if not latest_transcript: return f"No transcript content found for {symbol}." return latest_transcript except requests.exceptions.HTTPError as http_err: return f"HTTP error occurred: {http_err}" except requests.exceptions.RequestException as req_err: return f"Request error occurred: {req_err}" except Exception as err: return f"An unexpected error occurred: {err}" # Prompts summary_prompt = ( "You are a world-class financial analyst with extensive experience analyzing quarterly reports. " "Give me a comprehensive summary of the earnings report. Focus on the Strategic Insights and Key Financial Figures. " "Answer in extensive bullet points please." ) question_prompt = ( "You are a financial analyst with extensive experience analyzing quarterly reports. " "Read the earnings call transcript and earnings presentation report and generate 10 questions focusing on the strategic insights and financial figures. " "Ask questions that require precise answers and provide strategic insight into the company's financial and strategic performance, such as revenue growth, market trends, profit margins, and more. " "Only ask questions that can be answered using the provided document, without making any assumptions or inferences beyond the text. " "Please format the questions as a list with a simple '1. Question 1', '2. Question 2', etc. structure. " "Unless retrievable from the documents, don't ask questions which cannot be compared to previous periods." ) @cl.on_chat_start async def on_chat_start(): ticker_response = await cl.AskUserMessage( content=( "This tool is designed to analyze earnings call transcripts for publicly traded companies. " "Provide the company's ticker symbol, and the tool will fetch the latest earnings call transcript. " "It generates summaries and strategic due diligence. Ask your own questions afterwards. \n\n" "Please enter the ticker symbol for the company you want to analyze (e.g. MSFT):" ) ).send() ticker_symbol = ticker_response['content'].upper() msg = cl.Message(content=f"Retrieving earnings call transcript for {ticker_symbol}...") await msg.send() try: # Fetch the transcript using FMP API transcript_text = fetch_earnings_transcript(ticker_symbol) # Check if an error message was returned if transcript_text.startswith("No transcript") or \ transcript_text.startswith("HTTP error") or \ transcript_text.startswith("Request error") or \ transcript_text.startswith("An unexpected error occurred"): await cl.Message(content=transcript_text).send() return # Create a Document object with the transcript text document = Document(text=transcript_text, metadata={"company": ticker_symbol}) # Create index index = VectorStoreIndex.from_documents( [document], service_context=service_context ) # Store the index in the user session cl.user_session.set("index", index) # Generate summary query_engine = index.as_query_engine() summary_response = await cl.make_async(query_engine.query)(summary_prompt) await cl.Message(content=f"**Summary:**\n{summary_response}").send() # Generate questions questions_response = await cl.make_async(query_engine.query)(question_prompt) questions_format = str(questions_response).split('\n') relevant_questions = [question.strip() for question in questions_format if question.strip() and question.strip()[0].isdigit()] # Answer generated questions await cl.Message(content="Generated questions and answers:").send() for question in relevant_questions: response = await cl.make_async(query_engine.query)(question) await cl.Message(content=f"**{question}**\n{response}").send() msg.content = "Processing done. You can now ask more questions about the earnings call transcript!" await msg.update() except Exception as e: await cl.Message(content=f"An error occurred during processing: {str(e)}").send() @cl.on_message async def main(message: cl.Message): index = cl.user_session.get("index") if index is None: await cl.Message(content="Please provide a ticker symbol first before asking questions.").send() return query_engine = index.as_query_engine() response = await cl.make_async(query_engine.query)(message.content) response_message = cl.Message(content="") for token in str(response): await response_message.stream_token(token=token) await response_message.send()