当前位置：首页 > ds >正文

【python与生活】如何构建一个解读IPO招股书的算法？

ds 2025/7/21 20:07:36

在这里插入图片描述

构建一个基于Python的IPO招股书解读算法需要结合自然语言处理（NLP）技术和大型语言模型（LLM）。以下是一个完整的解决方案，使用LangChain框架和OpenAI的GPT模型：

import os
import re
import pandas as pd
from typing import List, Dict, Any, Optional
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.callbacks import get_openai_callback
import tiktokenclass IPOAnalyzer:def __init__(self, openai_api_key: str, model_name: str = "gpt-4"):"""初始化IPO招股书分析器"""os.environ["OPENAI_API_KEY"] = openai_api_keyself.embeddings = OpenAIEmbeddings()self.llm = ChatOpenAI(model_name=model_name, temperature=0)self.text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200,separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""])self.vectorstore = Noneself.qa_chain = Noneself.encoding = tiktoken.encoding_for_model(model_name)def load_pdf(self, pdf_path: str) -> List[str]:"""加载并分割PDF文档"""loader = PyPDFLoader(pdf_path)documents = loader.load()texts = self.text_splitter.split_documents(documents)print(f"文档已分割为{len(texts)}个文本块")return textsdef create_vectorstore(self, texts: List[str]) -> None:"""创建向量数据库"""self.vectorstore = FAISS.from_documents(texts, self.embeddings)print("向量数据库创建完成")def create_qa_chain(self) -> None:"""创建问答链"""prompt_template = """你是一位专业的金融分析师。请根据以下IPO招股书内容，回答问题。确保你的回答准确、客观，并基于提供的信息。如果信息不足，请明确指出。招股书内容：{context}问题：{question}回答："""PROMPT = PromptTemplate(template=prompt_template, input_variables=["context", "question"])self.qa_chain = RetrievalQA.from_chain_type(self.llm,retriever=self.vectorstore.as_retriever(),chain_type_kwargs={"prompt": PROMPT})def analyze_financials(self) -> Dict[str, Any]:"""分析财务数据"""questions = {"营收情况": "请总结最近三年的营业收入及其增长率","利润情况": "请总结最近三年的净利润及其增长率","毛利率": "请总结最近三年的毛利率及其变化趋势","资产负债": "请分析公司的资产负债结构和偿债能力","现金流": "请分析公司的现金流量状况"}results = {}with get_openai_callback() as cb:for category, question in questions.items():answer = self.qa_chain.run(question)results[category] = answerprint(f"完成{category}分析")print(f"财务分析总花费: {cb.total_cost}美元")return resultsdef analyze_business_model(self) -> Dict[str, Any]:"""分析商业模式"""questions = {"业务概述": "请描述公司的主要业务和产品","市场地位": "请分析公司在行业中的市场地位","竞争优势": "请总结公司的核心竞争优势","客户群体": "请描述公司的主要客户群体和销售渠道","商业模式": "请解释公司的商业模式和收入来源"}results = {}with get_openai_callback() as cb:for category, question in questions.items():answer = self.qa_chain.run(question)results[category] = answerprint(f"完成{category}分析")print(f"商业模式分析总花费: {cb.total_cost}美元")return resultsdef analyze_risk_factors(self) -> List[str]:"""分析风险因素"""question = "请列出公司在招股书中提到的主要风险因素，并简要说明"answer = self.qa_chain.run(question)# 尝试从回答中提取风险因素列表risk_pattern = r"[一二三四五六七八九十]、(.*?)(?:\n\n|$)"risks = re.findall(risk_pattern, answer, re.DOTALL)if not risks:# 如果正则表达式无法提取，直接返回原始回答risks = [answer]return risksdef generate_summary(self, financials: Dict, business: Dict, risks: List) -> str:"""生成综合分析报告"""prompt = f"""基于以下财务分析、商业模式分析和风险因素，为这家公司的IPO撰写一份全面的总结报告：财务分析：{financials}商业模式分析：{business}风险因素：{risks}请包括以下内容：1. 公司概况和核心业务2. 财务表现和关键指标3. 商业模式和竞争优势4. 主要风险和挑战5. 投资亮点和建议报告应客观、专业，约500-800字。"""with get_openai_callback() as cb:summary = self.qa_chain.run(prompt)print(f"总结报告生成完成，总花费: {cb.total_cost}美元")return summarydef run_analysis(self, pdf_path: str) -> Dict[str, Any]:"""执行完整的招股书分析流程"""# 加载并处理文档texts = self.load_pdf(pdf_path)self.create_vectorstore(texts)self.create_qa_chain()# 执行分析financials = self.analyze_financials()business = self.analyze_business_model()risks = self.analyze_risk_factors()summary = self.generate_summary(financials, business, risks)# 估算文档token数量sample_text = "".join([doc.page_content for doc in texts[:10]])sample_tokens = len(self.encoding.encode(sample_text))total_tokens = sample_tokens * (len(texts) / 10)return {"financial_analysis": financials,"business_analysis": business,"risk_factors": risks,"summary_report": summary,"document_stats": {"text_chunks": len(texts),"estimated_tokens": total_tokens}}# 使用示例
if __name__ == "__main__":# 设置你的OpenAI API密钥api_key = "your-openai-api-key"# 创建分析器实例analyzer = IPOAnalyzer(openai_api_key=api_key)# 执行分析（替换为实际招股书路径）analysis_results = analyzer.run_analysis("path/to/ipo_prospectus.pdf")# 保存结果with open("ipo_analysis_report.txt", "w", encoding="utf-8") as f:f.write("IPO招股书分析报告\n\n")f.write("=" * 50 + "\n\n")f.write("财务分析\n\n")for category, content in analysis_results["financial_analysis"].items():f.write(f"{category}:\n{content}\n\n")f.write("=" * 50 + "\n\n")f.write("商业模式分析\n\n")for category, content in analysis_results["business_analysis"].items():f.write(f"{category}:\n{content}\n\n")f.write("=" * 50 + "\n\n")f.write("风险因素\n\n")for i, risk in enumerate(analysis_results["risk_factors"], 1):f.write(f"{i}. {risk}\n\n")f.write("=" * 50 + "\n\n")f.write("综合分析报告\n\n")f.write(analysis_results["summary_report"])print("分析报告已保存至 ipo_analysis_report.txt")