DSPy-financial_analyst.py

import requests
import yfinance as yf
from yahooquery import Ticker

import warnings

# Ignore all warnings
warnings.filterwarnings('ignore')


class ThesisSignature(dspy.Signature):
  "A long and detailled investment thesis based on facts"
  news_analysis = dspy.InputField(desc="the recent news about the company")
  financial_analysis = dspy.InputField(desc="a report of the financial data of the company")
  investment_thesis = dspy.OutputField(desc="a detailled assessment of whether to buy the company's stock or not given news and financials. At least 5 paragraphs.")


class stock_analyst(dspy.Module):
  def __init__(self):
    super().__init__()
    self.stock_id_generator = dspy.Predict("company -> stock_ticker")
    self.news_analyzer = dspy.Predict("stock_news -> stock_recommendation_from_news")
    self.financial_analyzer = dspy.Predict("financial_data -> stock_recommendation_from_financial_data")
    self.investor = dspy.ChainOfThought(ThesisSignature)

  def get_company_news(self, company_name):
    params = {
        "engine": "google",
        "tbm": "nws",
        "q": company_name,
        "api_key": serpapi_key,
    }

    response = requests.get('https://serpapi.com/search', params=params)
    data = response.json()

    return data.get('news_results')

  def get_financial_statements(self, ticker):
    # Create a Ticker object
    company = Ticker(ticker)

    # Fetch data for Tesla
    company = yf.Ticker(ticker)

    # Get financials, balance sheet, and cash flow
    income_stmt = company.income_stmt
    balance_sheet = company.balance_sheet
    cashflow = company.cashflow
    recommendations = company.recommendations


    # Get key statistics
    keyStats = company.info

    # Extract and print valuation metrics
    market_cap = keyStats['marketCap']
    forward_pe = keyStats.get('forwardPE', 'N/A')  # Using .get() to avoid KeyError if the key doesn't exist
    price_to_book = keyStats.get('priceToBook', 'N/A')
    enterprise_value = keyStats['enterpriseValue']
    enterprise_to_revenue = keyStats.get('enterpriseToRevenue', 'N/A')
    enterprise_to_ebitda = keyStats.get('enterpriseToEbitda', 'N/A')

    return f"""balance_sheet: {balance_sheet}, income_statement: {income_stmt},
              cash_flow: {cashflow}, valuation_measures: 
              market cap: {market_cap}, forward_pe: {forward_pe},
              price to book: {price_to_book}, enterprise value: {enterprise_value},
              enterprise to revenue: {enterprise_to_revenue},
              enterprise to ebitda: {enterprise_to_ebitda}
            """


  def forward(self, company):
    ticker = self.stock_id_generator(company=company)
    news = self.get_company_news(company)
    financial_data = self.get_financial_statements(ticker.stock_ticker)
    news_analysis = self.news_analyzer(stock_news=str(news))
    financial_analysis = self.financial_analyzer(financial_data=financial_data)
    return self.investor(news_analysis=news_analysis.stock_recommendation_from_news,
                                   financial_analysis=financial_analysis.stock_recommendation_from_financial_data)


company = "Tesla"
stock_analyst = stock_analyst()
analysis = stock_analyst(company)

print(f"Investment thesis for {company}: {analysis.investment_thesis}")