"Service to recommend acquisitions" import json import logging import os from dotenv import load_dotenv from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from langchain_openai import ChatOpenAI from configs.config import OPENAI_API_KEY, OPENAI_MODEL_35 from models.pe_shops import FundGroups, PEShops from utils.chroma_db import ChromaDB as Query_ChromaDB from utils.dynamo_db import DynamoDB from utils.load_sic_codes import load_sic_codes from utils.url_parser import parsed_url load_dotenv() sic_codes_dict = load_sic_codes() class PERecommendation: "Class to recommend acquisitions" def __init__(self): # get API key from .env file self.openai_llm_4 = ChatOpenAI( api_key=OPENAI_API_KEY, model_name="gpt-4o", temperature=0.8, ) self.openai_llm_3 = ChatOpenAI( api_key=OPENAI_API_KEY, model_name=OPENAI_MODEL_35, temperature=0.2, ) def get_hypo_pe_shop( self, company_overview: str, product_description: str, open_ai_model=None, num_of_funds: int = 3, investment_criteria: dict = None, ): "Get hypothetical companies" open_ai_model = self.openai_llm_3 if open_ai_model is None else open_ai_model parser = JsonOutputParser(pydantic_object=PEShops) chat_template = ChatPromptTemplate.from_messages( [ ( "system", """You are an investment banking managing director. Being provided a company's overview and product description.""", ), ( "system", """Think about this step by step:\n 1) use the information to classify the company into a industry.\n 2) Become an investment banker for that industry and sector.\n 3) Take in the product description and anonymize it\n 4) Generate a synthetic investment thesis of a Private Equity shop who is interested in the company\n 5) Provide a minimum of {num_of_funds} PE shops that would be interested in the company\n """, ), ("system", "{json_structure}"), ("human", "{product_description}"), ("human", "Consider these investment criteria: {investment_criteria}"), ], ) chain = chat_template | open_ai_model | parser response = chain.invoke( { "num_of_funds": num_of_funds, "json_structure": parser.get_format_instructions(), "company overview": company_overview, "product_description": product_description, "investment_criteria": ( json.dumps(investment_criteria) if investment_criteria is not None else "" ), } ) return response def regroup_funds( self, target_company_description: str, funds: list[dict], open_ai_model=None ): "Regroup funds" open_ai_model = self.openai_llm_3 if open_ai_model is None else open_ai_model parser = JsonOutputParser(pydantic_object=FundGroups) previous_output = {"groups": []} batch_size = 10 for i in range(0, len(funds), batch_size): funds_for_this_run = funds[i : i + batch_size] funds_json = json.dumps(funds_for_this_run) chat_template = ChatPromptTemplate.from_messages( [ ( "system", """You are an investment banking managing director. Being provided with a set of company's product description. You are recommending PE shops that can acquire this company""", ), ( "system", """Think about this step by step:\n 1) Use the investment thesis to classify the funds into relevant groups.\n 2) Create a minimum of 2 groups and maximum of 5 with a recommendation of 3 groups. 3) Provide a rational for why the funds were grouped together especially with the provided description of target company. This should be clear and concise and should not be more than 3 sentences. 4) In the fund urls list, don't include the acquiring company. """, ), ("system", "{json_structure}"), ("human", "funds to group: ***\n {funds_json} \n ***"), ("human", "acquiring company: ***\n {target_company} \n ***"), # ( # "human", # "Add to the previous output: ***\n {previous_output} \n ***", # ), ], ) chain = chat_template | open_ai_model | parser response = chain.invoke( { "json_structure": parser.get_format_instructions(), "funds_json": funds_json, "target_company": target_company_description, # "previous_output": json.dumps(previous_output), } ) previous_output["groups"].extend(response["groups"]) return previous_output def get_recommendations(self, company_url: str, investment_criteria: dict = None): "Get PE recommendations for a company" query_chroma = Query_ChromaDB() # format the company url company_url = parsed_url(url=company_url, check_redirect=False).url # get company product description company = query_chroma.product_collection.get(ids=[company_url]) product_description = company["documents"][0] company_overview_docs = query_chroma.overview_collection.get(ids=[company_url]) company_overview = company_overview_docs["documents"][0] # get recommendations recommended_pe_shops = self.get_hypo_pe_shop( company_overview, product_description, num_of_funds=3, investment_criteria=investment_criteria, ) # retrieve companies from the db based on the recommended companies pe_shops_similar_to_recs = [] for pe_shop in recommended_pe_shops["PEShops"]: if pe_shop["investment_thesis"] == "": continue pe_shops_similar_to_recs.append( query_chroma.retrive_by_collection( pe_shop["investment_thesis"], query_chroma.fund_collection, n_results=20, ) ) # regroup companies funds_recommended = {} for group in pe_shops_similar_to_recs: for i, recommended_fund_url in enumerate(group["ids"][0]): # if the company is already in the list, skip funds_recommended[recommended_fund_url] = { "fund_url": recommended_fund_url, "investment_thesis": group["documents"][0][i], } regrouped_companies = self.regroup_funds( target_company_description=product_description, funds=list(funds_recommended.values()), ) return regrouped_companies def transform_recommender_output_for_front_end(self, recommender_output): "Transform the output of the recommender to the format expected by the frontend" get_item = DynamoDB() for group in recommender_output["groups"]: new_fund_list = [] for fund in group["funds"]: new_fund_list.append(get_item.get_item_for_fund_recommender(fund)) group["funds"] = new_fund_list return recommender_output if __name__ == "__main__": pe_shop_recs = PERecommendation() # print(acq_recommendor.get_public_comps("https://www.google.com") print( pe_shop_recs.get_recommendations( "https://bossinsights.com/", investment_criteria={ "industry": "Finance or Software or FinTech", "investment size": "1 million to 50 million", "geography": "North America", }, ), )