""" 金融研究报告生成器 整合数据获取、分析和文档生成功能，一键生成完整报告 """ import sys import argparse from datetime import datetime, timedelta from pathlib import Path from typing import Optional # 添加脚本目录到路径 sys.path.insert(0, str(Path(__file__).parent)) from docx import Document from docx.shared import Pt from docx.enum.text import WD_ALIGN_PARAGRAPH from data_fetcher import WindDataFetcher, fetch_stock_report_data from document_formatter import create_formatted_document, GBT9704Formatter from analysis_engine import perform_comprehensive_analysis class ReportGenerator: """ 金融研究报告生成器 """ def __init__(self): self.doc = None self.formatter = None self.analysis_results = None def generate_report(self, stock_code: str, stock_name: str = None, organization: str = "某某证券研究所", author: str = "", include_cover: bool = True, output_path: str = None) -> str: """ 生成完整的投资研究报告 Args: stock_code: 股票代码，如 '300866.SZ' stock_name: 股票名称（如不提供则自动获取） organization: 研究机构名称 author: 分析师姓名 include_cover: 是否包含封面 output_path: 输出文件路径 Returns: 生成的文件路径 """ print(f"开始生成 {stock_code} 的投资研究报告...") # 1. 获取数据 print("正在获取股票数据...") try: price_df, financial_data = fetch_stock_report_data(stock_code, days_back=90, weekly=True) except Exception as e: print(f"数据获取失败: {e}") raise # 如果未提供股票名称，从财务数据获取 if stock_name is None: stock_name = financial_data.get('sec_name', stock_code) print(f"获取到 {len(price_df)} 条周线数据") # 2. 执行分析 print("正在执行投资分析...") current_price = price_df['close'].iloc[-1] self.analysis_results = perform_comprehensive_analysis( price_df, financial_data, current_price ) # 3. 创建文档 print("正在生成Word文档...") self.doc, self.formatter = create_formatted_document() # 4. 添加封面（可选） if include_cover: self._add_cover_page(stock_name, organization) # 5. 添加免责声明 self._add_disclaimer() # 6. 添加核心提要 self._add_core_summary(stock_name) # 7. 添加投资评级 self._add_investment_rating(stock_name, current_price) # 8. 添加核心观点 self._add_core_views() # 9. 添加市场表现回顾 self._add_market_review(price_df) # 10. 添加页脚 self.formatter.add_page_number_footer() # 11. 保存文档 if output_path is None: date_str = datetime.now().strftime('%Y%m%d') output_path = f"{stock_name}_{stock_code.replace('.', '_')}_研究报告_{date_str}.docx" self.doc.save(output_path) print(f"报告已保存: {output_path}") return output_path def _add_cover_page(self, stock_name: str, organization: str): """添加封面页""" date_str = datetime.now().strftime('%Y年%m月%d日') self.formatter.add_cover_page( title=f"{stock_name}投资研究报告", organization=organization, date_str=date_str ) def _add_disclaimer(self): """添加免责声明""" self.formatter.add_heading('免责声明', level=1) self.formatter.add_paragraph( "本报告所载资料的来源及观点皆为公开信息，但并不能保证其准确性和完整性。本报告仅供参考，" "不构成任何投资建议或承诺，投资者应审慎决策，独立判断，自行承担投资风险。" ) self.formatter.add_paragraph( "本报告版权归本公司所有，未经书面许可，任何机构和个人不得以任何形式翻版、复制、刊登、" "发表或引用。如征得本公司同意进行引用、刊发的，需在允许的范围内使用，并注明出处为" "'某某证券研究所'，且不得对本报告进行任何有悖原意的引用、删节和修改。" ) def _add_core_summary(self, stock_name: str): """添加核心提要""" self.formatter.add_heading('核心提要', level=1) summary_points = self.analysis_results['summary_points'] for i, point in enumerate(summary_points, 1): # 添加项目符号 paragraph = self.doc.add_paragraph() run = paragraph.add_run(f"{i}. {point}") self.formatter.set_font(run, '仿宋', 14) paragraph.paragraph_format.line_spacing = Pt(28) paragraph.paragraph_format.line_spacing_rule = 2 paragraph.paragraph_format.space_before = Pt(0) paragraph.paragraph_format.space_after = Pt(0) def _add_investment_rating(self, stock_name: str, current_price: float): """添加投资评级""" self.formatter.add_heading('一、投资评级', level=1) # 表格数据 recommendation = self.analysis_results['recommendation'] data = [[ recommendation.rating.value, f"{recommendation.target_price:.2f}元", (datetime.now() + timedelta(days=365)).strftime('%Y年%m月%d日'), recommendation.risk_level.value ]] self.formatter.add_three_line_table( data, ['投资建议', '目标价格', '有效期', '风险等级'], font_name='仿宋', size=12 ) # 添加说明文字 self.formatter.add_paragraph( f"当前股价：{current_price:.2f}元。" f"目标价较当前价格{'上涨' if recommendation.target_price > current_price else '下跌'}" f"{abs((recommendation.target_price / current_price - 1) * 100):.2f}%。" ) def _add_core_views(self): """添加核心观点""" self.formatter.add_heading('二、核心观点', level=1) # 基本面分析 fundamental = self.analysis_results['fundamental'] self.formatter.add_heading('（一）基本面分析', level=2) self.formatter.add_paragraph( f"1. 盈利能力：公司最新净资产收益率（ROE）为{fundamental.roe}%，" f"{fundamental.roe_comment}。" ) self.formatter.add_paragraph( f"2. 估值水平：当前市盈率（TTM）为{fundamental.pe_ttm}倍，" f"市净率（LF）为{fundamental.pb_lf}倍。{fundamental.valuation_comment}。" ) self.formatter.add_paragraph(f"3. 结论：{fundamental.fundamental_summary}") # 技术面分析 technical = self.analysis_results['technical'] self.formatter.add_heading('（二）技术面分析', level=2) self.formatter.add_paragraph( f"1. 价格趋势：近期股价呈现{technical.trend}态势，" f"{technical.ma_status}。" ) self.formatter.add_paragraph( f"2. 支撑与阻力：近期支撑位{technical.support_level}元，" f"阻力位{technical.resistance_level}元。" ) self.formatter.add_paragraph( f"3. 波动率：近期年化波动率为{technical.volatility}%，" f"{'属于高波动品种' if technical.volatility > 40 else '波动适中' if technical.volatility > 25 else '波动较低'}。" ) # 投资建议 recommendation = self.analysis_results['recommendation'] self.formatter.add_heading('（三）投资建议', level=2) self.formatter.add_paragraph(recommendation.recommendation_text) self.formatter.add_paragraph("主要投资逻辑：") for factor in recommendation.key_factors: paragraph = self.doc.add_paragraph() run = paragraph.add_run(f"• {factor}") self.formatter.set_font(run, '仿宋', 14) paragraph.paragraph_format.line_spacing = Pt(28) paragraph.paragraph_format.line_spacing_rule = 2 paragraph.paragraph_format.first_line_indent = Pt(28) def _add_market_review(self, price_df): """添加市场表现回顾""" self.formatter.add_heading('三、市场表现回顾', level=1) # 价格走势分析 self.formatter.add_heading('（一）价格走势分析', level=2) price_stats = self.analysis_results['price_stats'] self.formatter.add_paragraph( f"近期股价最高达到{price_stats['highest_price']}元，" f"最低下探至{price_stats['lowest_price']}元，" f"区间累计涨跌幅为{price_stats['total_change']:.2f}%。" f"最新收盘价为{price_stats['latest_price']}元，" f"区间平均价格为{price_stats['avg_price']}元。" ) # K线数据表 self.formatter.add_heading('（二）K线数据', level=2) # 准备表格数据（最近13周） recent_data = price_df.tail(13).copy() table_data = [] for _, row in recent_data.iterrows(): date_str = row['date'].strftime('%Y-%m-%d') if hasattr(row['date'], 'strftime') else str(row['date'])[:10] table_data.append([ date_str, f"{row['open']:.2f}", f"{row['high']:.2f}", f"{row['low']:.2f}", f"{row['close']:.2f}", f"{row.get('pct_chg', 0):.2f}%", f"{row.get('volume', 0) / 10000:.0f}" ]) self.formatter.add_three_line_table( table_data, ['日期', '开盘价', '最高价', '最低价', '收盘价', '涨跌幅', '成交量(万股)'], font_name='仿宋', size=12 ) # 数据来源 self.formatter.add_data_source_footnote() def main(): """主函数""" parser = argparse.ArgumentParser(description='生成金融投资研究报告') parser.add_argument('stock_code', help='股票代码，如 300866.SZ 或 000001') parser.add_argument('--name', '-n', help='股票名称') parser.add_argument('--org', '-o', default='某某证券研究所', help='研究机构名称') parser.add_argument('--author', '-a', default='', help='分析师姓名') parser.add_argument('--no-cover', action='store_true', help='不包含封面') parser.add_argument('--output', '-p', help='输出文件路径') args = parser.parse_args() # 标准化股票代码 stock_code = args.stock_code if '.' not in stock_code: if stock_code.startswith('6'): stock_code = f"{stock_code}.SH" else: stock_code = f"{stock_code}.SZ" generator = ReportGenerator() try: output_path = generator.generate_report( stock_code=stock_code, stock_name=args.name, organization=args.org, author=args.author, include_cover=not args.no_cover, output_path=args.output ) print(f"\n报告生成成功: {output_path}") except Exception as e: print(f"\n报告生成失败: {e}") import traceback traceback.print_exc() sys.exit(1) if __name__ == '__main__': main()