"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.comapny_groups import CompanyGroups from models.companies import Companies from utils.chroma_db import ChromaDB as Query_ChromaDB from utils.dynamo_db import DynamoDB from utils.load_sic_codes import load_sic_codes load_dotenv() OPENAI_API_KEY = os.environ["OPENAI_API_KEY"] sic_codes_dict = load_sic_codes() class AcquisitionsRecommender: "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_acq_companies( self, product_description: str, open_ai_model=None, num_of_companies: int = 3, acq_direction: str = "horizontal", ): "Get hypothetical companies" open_ai_model = self.openai_llm_3 if open_ai_model is None else open_ai_model parser = JsonOutputParser(pydantic_object=Companies) chat_template = ChatPromptTemplate.from_messages( [ ( "system", """You are an investment banking managing director. Being provided a company's products description.""", ), ( "system", """Think about this step by step:\n 1) Use the product description to classify the company into an industry and sector.\n 2) Become an investment banker for that industry and sector.\n 3) Take in the product description and anonymize it\n 5) Generate a syntenic description of a company that would be interest\n 4) Provide 3 example companies with 5 bullet points describing the company's business\n 5) Do not include rational for the provided company """, ), ("system", "{json_structure}"), ("human", "{product_description}"), ("human", "{pitch_direction}"), ( "system", "You should use this dictionary for SIC codes {sic_codes_dict}", ), ], ) chain = chat_template | open_ai_model | parser response = chain.invoke( { "num_of_companies": num_of_companies, "json_structure": parser.get_format_instructions(), "product_description": product_description, "pitch_direction": f""" Write product descriptions of companies that would be good {acq_direction} acquisition targets for this company. """, "sic_codes_dict": json.dumps(sic_codes_dict), } ) return response def regroup_companies( self, acq_company_description: str, companies: list[dict], open_ai_model=None ): "Regroup companies" open_ai_model = self.openai_llm_3 if open_ai_model is None else open_ai_model parser = JsonOutputParser(pydantic_object=CompanyGroups) companies_json = json.dumps(companies) chat_template = ChatPromptTemplate.from_messages( [ ( "system", """You are an investment banking managing director. Being provided with a set of company's url and product descriptions.""", ), ( "system", """Think about this step by step:\n 1) Use the product description to classify the companies 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 companies were grouped together especially with the provided description of acquiring company. This should be clear and concise and should not be more than 3 sentences. 4) In the company urls list, don't include the acquiring company. """, ), ("system", "{json_structure}"), ("human", "companies to group: ***\n {companies_json} \n ***"), ("human", "acquiring company: ***\n {acq_company} \n ***"), ], ) chain = chat_template | open_ai_model | parser response = chain.invoke( { "json_structure": parser.get_format_instructions(), "companies_json": companies_json, "acq_company": acq_company_description, } ) return response def get_recommendations( self, company_url: str, acq_direction: str = "horizontal", public: bool = None ): "Get recommendations for a company" query_chroma = Query_ChromaDB() # get company product description company = query_chroma.product_collection.get(ids=[company_url]) product_description = company["documents"][0] # get recommendations recommended_companies = self.get_hypo_acq_companies( product_description, self.openai_llm_3, num_of_companies=3, acq_direction=acq_direction, ) if public is True: public_filter = {"$eq": True} elif public is False: public_filter = {"$eq": False} else: public_filter = None industry_filter = {"$or": []} for rec_company in recommended_companies["companies"]: try: # if the SIC code is not in the dictionary, skip industry_filter["$or"].append( { "$and": [ { "industry": { "$eq": sic_codes_dict[int(rec_company["SIC_code"])][ "industry" ] } }, { "sector": { "$eq": sic_codes_dict[int(rec_company["SIC_code"])][ "sector" ] } }, ] } ) except KeyError: logging.error( f"SIC code {rec_company['SIC_code']} not found in the dictionary" ) continue where = query_chroma.create_where_filter(industry_filter, public=public_filter) # retrieve companies from the db based on the recommended companies companies_similar_to_recommended_group = [] for company in recommended_companies["companies"]: companies_similar_to_recommended_group.append( query_chroma.retrive_by_collection( company["product_descriptions"], query_chroma.product_collection, where=where, where_document=None, ) ) # regroup companies companies_recommended = {} for group in companies_similar_to_recommended_group: for i, recommended_company_url in enumerate(group["ids"][0]): # if the company is the same as the company we are trying to find recommendations for, skip if company_url == recommended_company_url: continue # if the company is already in the list, skip companies_recommended[recommended_company_url] = { "company_url": recommended_company_url, "product_description": group["documents"][0][i], } regrouped_companies = self.regroup_companies( acq_company_description=product_description, companies=list(companies_recommended.values()), open_ai_model=self.openai_llm_3, ) return regrouped_companies def get_public_comps(self, company_url: str): "Get recommendations for a company" query_chroma = Query_ChromaDB() # get company product description company = query_chroma.product_collection.get(ids=[company_url]) product_description = company["documents"][0] where = query_chroma.create_where_filter( {"industry": {"$eq": company["metadatas"][0]["industry"]}}, public={"$eq": True}, ) # retrieve companies from the db based on the recommended companies companies_similar_to_recommended_group = [] companies_similar_to_recommended_group.append( query_chroma.retrive_by_collection( [product_description], query_chroma.product_collection, where=where, where_document=None, ) ) # regroup companies companies_recommended = {} for group in companies_similar_to_recommended_group: for i, recommended_company_url in enumerate(group["ids"][0]): # if the company is the same as the company # we are trying to find recommendations for, skip if company_url == recommended_company_url: continue # if the company is already in the list, skip companies_recommended[recommended_company_url] = { "company_url": recommended_company_url, "product_description": group["documents"][0][i], } regrouped_companies = self.regroup_companies( acq_company_description=product_description, companies=list(companies_recommended.values()), open_ai_model=self.openai_llm_3, ) 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_company_list = [] for company in group["companies"]: new_company_list.append(get_item.get_item_for_recommender(company)) group["companies"] = new_company_list return recommender_output if __name__ == "__main__": import sys sys.path.append(r".") acq_recommendor = AcquisitionsRecommender() # print(acq_recommendor.get_public_comps("https://www.google.com")) print(acq_recommendor.get_recommendations("https://www.google.com")) print(acq_recommendor.get_recommendations("https://www.unity.com")) print(acq_recommendor.get_recommendations("https://www.lyft.com"))