stock-analyst-dspy.py

import requests
from yahooquery import Ticker
import dspy
from config import OPENAI_API_KEY, SERPAPI_API_KEY

llm = dspy.OpenAI(model='gpt-3.5-turbo',api_key=OPENAI_API_KEY, max_tokens=2000)

dspy.settings.configure(lm=llm)


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 -> crafted_stock_recommendation_from_news")
        self.financial_analyzer = dspy.Predict("financial_data -> crafted_stock_recommendation_from_financial_data")
        self.investor = dspy.ChainOfThought("news_analysis, financial_analysis -> detailed_investment_thesis")

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

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

        return f"news: {data.get('news_results')}"

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

        # Get financial data
        balance_sheet = company.balance_sheet().to_string()
        cash_flow = company.cash_flow(trailing=False).to_string()
        income_statement = company.income_statement().to_string()
        valuation_measures = str(company.valuation_measures)

        input_string = (
            f"""balance_sheet: {balance_sheet}, income_statement: {income_statement},
            cash_flow: {cash_flow}, valuation_measures: {valuation_measures}"""
            )
        truncated_string = input_string[:2000]

        return truncated_string

    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=news)
        financial_analysis = self.financial_analyzer(financial_data=financial_data)
        return self.investor(news_analysis=news_analysis.crafted_stock_recommendation_from_news,
                                    financial_analysis=financial_analysis.crafted_stock_recommendation_from_financial_data)


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

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