import os import traceback from reportlab.lib import colors from reportlab.lib import pagesizes from reportlab.platypus import ( SimpleDocTemplate, Frame, Paragraph, Image, PageTemplate, FrameBreak, Spacer, Table, TableStyle, NextPageTemplate, PageBreak, ) from reportlab.lib.units import inch from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.enums import TA_CENTER, TA_JUSTIFY, TA_LEFT from ..data_source import FMPUtils, YFinanceUtils from .analyzer import ReportAnalysisUtils from typing import Annotated class ReportLabUtils: def build_annual_report( ticker_symbol: Annotated[str, "ticker symbol"], save_path: Annotated[str, "path to save the annual report pdf"], income_summarization: Annotated[ str, "a paragraph of text: the company's income summarization from its financial report", ], business_highlights: Annotated[ str, "a paragraph of text: the company's business highlights from its financial report", ], company_description: Annotated[ str, "a paragraph of text: the company's description and current situation from its financial report", ], risk_assessment: Annotated[ str, "a paragraph of text: the company's risk assessment from its financial report", ], share_performance_image_path: Annotated[ str, "path to the share performance image" ], pe_eps_performance_image_path: Annotated[ str, "path to the PE and EPS performance image" ], filing_date: Annotated[str, "filing date of the analyzed financial report"], ) -> str: """ Aggregate a company's income summarization, business highlights, company description, risk assessment and share performance, PE & EPS performance charts all into a PDF report. """ try: # 2. 创建PDF并插入图像 # 页面设置 page_width, page_height = pagesizes.A4 left_column_width = page_width * 2 / 3 right_column_width = page_width - left_column_width margin = 4 # 创建PDF文档路径 pdf_path = ( os.path.join(save_path, f"{ticker_symbol}_report.pdf") if os.path.isdir(save_path) else save_path ) os.makedirs(os.path.dirname(pdf_path), exist_ok=True) doc = SimpleDocTemplate(pdf_path, pagesize=pagesizes.A4) # 定义两个栏位的Frame frame_left = Frame( margin, margin, left_column_width - margin * 2, page_height - margin * 2, id="left", ) frame_right = Frame( left_column_width, margin, right_column_width - margin * 2, page_height - margin * 2, id="right", ) # single_frame = Frame(margin, margin, page_width-margin*2, page_height-margin*2, id='single') # single_column_layout = PageTemplate(id='OneCol', frames=[single_frame]) left_column_width_p2 = (page_width - margin * 3) // 2 right_column_width_p2 = left_column_width_p2 frame_left_p2 = Frame( margin, margin, left_column_width_p2 - margin * 2, page_height - margin * 2, id="left", ) frame_right_p2 = Frame( left_column_width_p2, margin, right_column_width_p2 - margin * 2, page_height - margin * 2, id="right", ) # 创建PageTemplate，并添加到文档 page_template = PageTemplate( id="TwoColumns", frames=[frame_left, frame_right] ) page_template_p2 = PageTemplate( id="TwoColumns_p2", frames=[frame_left_p2, frame_right_p2] ) # Define single column Frame single_frame = Frame( margin, margin, page_width - 2 * margin, page_height - 2 * margin, id="single", ) # Create a PageTemplate with a single column single_column_layout = PageTemplate(id="OneCol", frames=[single_frame]) doc.addPageTemplates( [page_template, single_column_layout, page_template_p2] ) styles = getSampleStyleSheet() # 自定义样式 custom_style = ParagraphStyle( name="Custom", parent=styles["Normal"], fontName="Helvetica", fontSize=10, # leading=15, alignment=TA_JUSTIFY, ) title_style = ParagraphStyle( name="TitleCustom", parent=styles["Title"], fontName="Helvetica-Bold", fontSize=16, leading=20, alignment=TA_LEFT, spaceAfter=10, ) subtitle_style = ParagraphStyle( name="Subtitle", parent=styles["Heading2"], fontName="Helvetica-Bold", fontSize=14, leading=12, alignment=TA_LEFT, spaceAfter=6, ) table_style2 = TableStyle( [ ("BACKGROUND", (0, 0), (-1, -1), colors.white), ("BACKGROUND", (0, 0), (-1, 0), colors.white), ("FONT", (0, 0), (-1, -1), "Helvetica", 7), ("FONT", (0, 0), (-1, 0), "Helvetica-Bold", 14), ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), # 所有单元格左对齐 ("ALIGN", (0, 0), (-1, -1), "LEFT"), # 标题栏下方添加横线 ("LINEBELOW", (0, 0), (-1, 0), 2, colors.black), # 表格最下方添加横线 ("LINEBELOW", (0, -1), (-1, -1), 2, colors.black), ] ) name = YFinanceUtils.get_stock_info(ticker_symbol)["shortName"] # 准备左栏和右栏内容 content = [] # 标题 content.append( Paragraph( f"Equity Research Report: {name}", title_style, ) ) # 子标题 content.append(Paragraph("Income Summarization", subtitle_style)) content.append(Paragraph(income_summarization, custom_style)) content.append(Paragraph("Business Highlights", subtitle_style)) content.append(Paragraph(business_highlights, custom_style)) content.append(Paragraph("Company Situation", subtitle_style)) content.append(Paragraph(company_description, custom_style)) content.append(Paragraph("Risk Assessment", subtitle_style)) content.append(Paragraph(risk_assessment, custom_style)) # content.append(Paragraph("Summarization", subtitle_style)) df = FMPUtils.get_financial_metrics(ticker_symbol, years=5) df.reset_index(inplace=True) currency = YFinanceUtils.get_stock_info(ticker_symbol)["currency"] df.rename(columns={"index": f"FY ({currency} mn)"}, inplace=True) table_data = [["Financial Metrics"]] table_data += [df.columns.to_list()] + df.values.tolist() col_widths = [(left_column_width - margin * 4) / df.shape[1]] * df.shape[1] table = Table(table_data, colWidths=col_widths) table.setStyle(table_style2) content.append(table) content.append(FrameBreak()) # 用于从左栏跳到右栏 table_style = TableStyle( [ ("BACKGROUND", (0, 0), (-1, -1), colors.white), ("BACKGROUND", (0, 0), (-1, 0), colors.white), ("FONT", (0, 0), (-1, -1), "Helvetica", 8), ("FONT", (0, 0), (-1, 0), "Helvetica-Bold", 12), ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), # 第一列左对齐 ("ALIGN", (0, 1), (0, -1), "LEFT"), # 第二列右对齐 ("ALIGN", (1, 1), (1, -1), "RIGHT"), # 标题栏下方添加横线 ("LINEBELOW", (0, 0), (-1, 0), 2, colors.black), ] ) full_length = right_column_width - 2 * margin data = [ ["FinRobot"], ["https://ai4finance.org/"], ["https://github.com/AI4Finance-Foundation/FinRobot"], [f"Report date: {filing_date}"], ] col_widths = [full_length] table = Table(data, colWidths=col_widths) table.setStyle(table_style) content.append(table) # content.append(Paragraph("", custom_style)) content.append(Spacer(1, 0.15 * inch)) key_data = ReportAnalysisUtils.get_key_data(ticker_symbol, filing_date) # 表格数据 data = [["Key data", ""]] data += [[k, v] for k, v in key_data.items()] col_widths = [full_length // 3 * 2, full_length // 3] table = Table(data, colWidths=col_widths) table.setStyle(table_style) content.append(table) # 将Matplotlib图像添加到右栏 # 历史股价 data = [["Share Performance"]] col_widths = [full_length] table = Table(data, colWidths=col_widths) table.setStyle(table_style) content.append(table) plot_path = share_performance_image_path width = right_column_width height = width // 2 content.append(Image(plot_path, width=width, height=height)) # 历史PE和EPS data = [["PE & EPS"]] col_widths = [full_length] table = Table(data, colWidths=col_widths) table.setStyle(table_style) content.append(table) plot_path = pe_eps_performance_image_path width = right_column_width height = width // 2 content.append(Image(plot_path, width=width, height=height)) # # 开始新的一页 # content.append(NextPageTemplate("OneCol")) # content.append(PageBreak()) # def add_table(df, title): # df = df.applymap(lambda x: "{:.2f}".format(x) if isinstance(x, float) else x) # # df.columns = [col.strftime('%Y') for col in df.columns] # # df.reset_index(inplace=True) # # currency = ra.info['currency'] # df.rename(columns={"index": "segment"}, inplace=True) # table_data = [[title]] # table_data += [df.columns.to_list()] + df.values.tolist() # table = Table(table_data) # table.setStyle(table_style2) # num_columns = len(df.columns) # column_width = (page_width - 4 * margin) / (num_columns + 1) # first_column_witdh = column_width * 2 # table._argW = [first_column_witdh] + [column_width] * (num_columns - 1) # content.append(table) # content.append(Spacer(1, 0.15 * inch)) # if os.path.exists(f"{ra.project_dir}/outer_resource/"): # Revenue10Q = pd.read_csv( # f"{ra.project_dir}/outer_resource/Revenue10Q.csv", # ) # # del Revenue10K['FY2018'] # # del Revenue10K['FY2019'] # add_table(Revenue10Q, "Revenue") # Ratio10Q = pd.read_csv( # f"{ra.project_dir}/outer_resource/Ratio10Q.csv", # ) # # del Ratio10K['FY2018'] # # del Ratio10K['FY2019'] # add_table(Ratio10Q, "Ratio") # Yoy10Q = pd.read_csv( # f"{ra.project_dir}/outer_resource/Yoy10Q.csv", # ) # # del Yoy10K['FY2018'] # # del Yoy10K['FY2019'] # add_table(Yoy10Q, "Yoy") # plot_path = os.path.join(f"{ra.project_dir}/outer_resource/", "segment.png") # width = page_width - 2 * margin # height = width * 3 // 5 # content.append(Image(plot_path, width=width, height=height)) # # 第二页及之后内容，使用单栏布局 # df = ra.get_income_stmt() # df = df[df.columns[:3]] # def convert_if_money(value): # if np.abs(value) >= 1000000: # return value / 1000000 # else: # return value # # 应用转换函数到DataFrame的每列 # df = df.applymap(convert_if_money) # df.columns = [col.strftime('%Y') for col in df.columns] # df.reset_index(inplace=True) # currency = ra.info['currency'] # df.rename(columns={'index': f'FY ({currency} mn)'}, inplace=True) # 可选：重命名索引列为“序号” # table_data = [["Income Statement"]] # table_data += [df.columns.to_list()] + df.values.tolist() # table = Table(table_data) # table.setStyle(table_style2) # content.append(table) # content.append(FrameBreak()) # 用于从左栏跳到右栏 # df = ra.get_cash_flow() # df = df[df.columns[:3]] # df = df.applymap(convert_if_money) # df.columns = [col.strftime('%Y') for col in df.columns] # df.reset_index(inplace=True) # currency = ra.info['currency'] # df.rename(columns={'index': f'FY ({currency} mn)'}, inplace=True) # 可选：重命名索引列为“序号” # table_data = [["Cash Flow Sheet"]] # table_data += [df.columns.to_list()] + df.values.tolist() # table = Table(table_data) # table.setStyle(table_style2) # content.append(table) # # content.append(Paragraph('This is a single column on the second page', custom_style)) # # content.append(Spacer(1, 0.2*inch)) # # content.append(Paragraph('More content in the single column.', custom_style)) # 构建PDF文档 doc.build(content) return "Annual report generated successfully." except Exception: return traceback.format_exc()