Spaces:
Sleeping
Sleeping
| 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() | |