create_volume_energy_pdf.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
市场成交量能分析独立PDF报告生成器
功能：调用成交量能分析模块，生成专业的PDF分析报告
"""

import os
import sys
import json
from datetime import datetime
from typing import Dict, Any
import logging

# 导入PDF生成相关模块
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import inch
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Image, Table, TableStyle
from reportlab.lib import colors
from reportlab.lib.enums import TA_CENTER, TA_LEFT, TA_JUSTIFY

# 导入自定义分析模块
from analyze_volume_energy import VolumeEnergyAnalyzer, create_volume_energy_charts

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 设置中文字体支持
try:
    # 尝试注册中文字体
    import platform
    system = platform.system()
    
    if system == "Windows":
        # Windows系统字体路径
        font_paths = [
            'C:/Windows/Fonts/simhei.ttf',
            'C:/Windows/Fonts/simsun.ttc', 
            'C:/Windows/Fonts/msyh.ttc',
            'C:/Windows/Fonts/simkai.ttf'
        ]
    else:
        # Linux/Mac系统字体路径
        font_paths = [
            '/usr/share/fonts/truetype/arphic/ukai.ttc',
            '/System/Library/Fonts/PingFang.ttc',
            '/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf'
        ]
    
    CHINESE_FONT = 'Helvetica'  # 默认字体
    
    for font_path in font_paths:
        if os.path.exists(font_path):
            try:
                from reportlab.pdfbase import pdfmetrics
                from reportlab.pdfbase.ttfonts import TTFont
                pdfmetrics.registerFont(TTFont('SimHei', font_path))
                CHINESE_FONT = 'SimHei'
                logger.info(f"成功加载中文字体: {font_path}")
                break
            except Exception as e:
                logger.warning(f"尝试加载字体失败 {font_path}: {e}")
                continue
    
    if CHINESE_FONT == 'Helvetica':
        logger.warning("未找到中文字体，将使用默认字体，可能出现中文显示问题")

except Exception as e:
    logger.error(f"字体设置失败: {e}")
    CHINESE_FONT = 'Helvetica'


class VolumeEnergyPDFGenerator:
    """市场成交量能分析PDF报告生成器"""
    
    def __init__(self):
        """初始化PDF生成器"""
        self.setup_styles()
    
    def setup_styles(self):
        """设置PDF样式"""
        self.styles = getSampleStyleSheet()
        
        # 标题样式
        self.title_style = ParagraphStyle(
            'CustomTitle',
            parent=self.styles['Heading1'],
            fontSize=20,
            spaceAfter=30,
            alignment=TA_CENTER,
            fontName=CHINESE_FONT
        )
        
        # 副标题样式
        self.subtitle_style = ParagraphStyle(
            'CustomSubtitle',
            parent=self.styles['Heading2'],
            fontSize=16,
            spaceAfter=20,
            spaceBefore=15,
            textColor=colors.darkblue,
            fontName=CHINESE_FONT
        )
        
        # 正文样式
        self.normal_style = ParagraphStyle(
            'CustomNormal',
            parent=self.styles['Normal'],
            fontSize=12,
            spaceAfter=12,
            alignment=TA_JUSTIFY,
            leading=18,
            fontName=CHINESE_FONT
        )
        
        # 重要信息样式
        self.highlight_style = ParagraphStyle(
            'Highlight',
            parent=self.styles['Normal'],
            fontSize=12,
            spaceAfter=12,
            textColor=colors.darkred,
            backColor=colors.lightgrey,
            borderColor=colors.grey,
            borderWidth=1,
            borderPadding=8,
            fontName=CHINESE_FONT
        )
        
        # 图表说明样式
        self.caption_style = ParagraphStyle(
            'Caption',
            parent=self.styles['Normal'],
            fontSize=10,
            spaceAfter=10,
            spaceBefore=5,
            alignment=TA_CENTER,
            textColor=colors.grey,
            fontName=CHINESE_FONT
        )
    
    def create_comprehensive_report(self, trade_date: str, output_dir: str = "reports") -> str:
        """创建综合的成交量能分析PDF报告"""
        try:
            logger.info(f"开始生成{trade_date}的成交量能分析PDF报告...")
            
            # 确保输出目录存在
            os.makedirs(output_dir, exist_ok=True)
            
            # 执行分析
            analyzer = VolumeEnergyAnalyzer()
            analysis_result = analyzer.generate_comprehensive_analysis(trade_date)
            
            if not analysis_result:
                logger.error("分析失败，无法生成报告")
                return ""
            
            # 生成图表
            chart_files = create_volume_energy_charts(analysis_result)
            
            # 创建PDF文件 - 添加时间戳避免冲突
            timestamp = datetime.now().strftime('%H%M%S')
            pdf_filename = f"Volume_Energy_Analysis_{trade_date}_{timestamp}.pdf"
            pdf_path = os.path.join(output_dir, pdf_filename)
            
            # 创建PDF文档
            doc = SimpleDocTemplate(
                pdf_path,
                pagesize=A4,
                rightMargin=72,
                leftMargin=72,
                topMargin=72,
                bottomMargin=18
            )
            
            # 构建内容
            content = []
            
            # 添加标题
            content.append(Paragraph("市场成交量能深度分析报告", self.title_style))
            content.append(Paragraph(f"分析日期：{self._format_date(trade_date)}", self.normal_style))
            content.append(Paragraph(f"报告生成时间：{analysis_result.get('analysis_time', datetime.now().strftime('%Y-%m-%d %H:%M:%S'))}", self.normal_style))
            content.append(Spacer(1, 20))
            
            # 1. 执行摘要
            content.append(Paragraph("一、执行摘要", self.subtitle_style))
            
            # 综合评级表格
            summary_data = [
                ['评价项目', '结果', '说明'],
                ['综合评分', f"{analysis_result.get('comprehensive_score', 0)}/100", ''],
                ['总体评级', analysis_result.get('overall_rating', '未知'), analysis_result.get('rating_desc', '')],
                ['分析总结', '', analysis_result.get('summary', '')]
            ]
            
            summary_table = Table(summary_data, colWidths=[2*inch, 1.5*inch, 3*inch])
            summary_table.setStyle(TableStyle([
                ('BACKGROUND', (0, 0), (-1, 0), colors.grey),
                ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),
                ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
                ('FONTNAME', (0, 0), (-1, -1), CHINESE_FONT),
                ('FONTSIZE', (0, 0), (-1, -1), 10),
                ('BOTTOMPADDING', (0, 0), (-1, 0), 12),
                ('BACKGROUND', (0, 1), (-1, -1), colors.beige),
                ('SPAN', (1, 3), (2, 3)),
                ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'),
                ('GRID', (0, 0), (-1, -1), 1, colors.black),
            ]))
            
            content.append(summary_table)
            content.append(Spacer(1, 20))
            
            # 2. 市场成交额趋势分析
            content.append(Paragraph("二、市场成交额趋势分析", self.subtitle_style))
            
            volume_trend = analysis_result.get('volume_trend', {})
            if volume_trend:
                # 添加趋势分析图表
                trend_chart = self._find_chart(chart_files, 'volume_trend')
                if trend_chart and os.path.exists(trend_chart):
                    content.append(Image(trend_chart, width=6*inch, height=4.5*inch))
                    content.append(Paragraph("图1: 市场成交额趋势分析", self.caption_style))
                    content.append(Spacer(1, 15))
                
                # 趋势分析详情
                trend_text = f"""
                <b>当前成交状况：</b><br/>
                • 最新成交额：{volume_trend.get('latest_amount', 0)}亿元<br/>
                • 5日均值：{volume_trend.get('ma5_amount', 0)}亿元<br/>
                • 20日均值：{volume_trend.get('ma20_amount', 0)}亿元<br/>
                • 相对20日均线比值：{volume_trend.get('ma20_ratio', 0)}<br/>
                <br/>
                <b>量能水平评估：</b><br/>
                • 量能等级：{volume_trend.get('volume_level', '未知')}<br/>
                • 趋势描述：{volume_trend.get('trend_desc', '未知')}<br/>
                • 动量指标：{volume_trend.get('momentum', 0)}<br/>
                """
                
                content.append(Paragraph(trend_text, self.normal_style))
                content.append(Spacer(1, 15))
            
            content.append(Spacer(1, 20))
            
            # 3. 量价配合度分析
            content.append(Paragraph("三、量价配合度分析", self.subtitle_style))
            
            harmony_analysis = analysis_result.get('harmony_analysis', {})
            if harmony_analysis:
                # 添加量价配合度图表
                harmony_chart = self._find_chart(chart_files, 'volume_price_harmony')
                if harmony_chart and os.path.exists(harmony_chart):
                    content.append(Image(harmony_chart, width=6*inch, height=3*inch))
                    content.append(Paragraph("图2: 量价配合度分析", self.caption_style))
                    content.append(Spacer(1, 15))
                
                # 量价分析详情
                harmony_text = f"""
                <b>量价配合度评估：</b><br/>
                • 配合度评分：{harmony_analysis.get('harmony_score', 0)}/100<br/>
                • 配合度等级：{harmony_analysis.get('harmony_level', '未知')}<br/>
                • 当前信号：{harmony_analysis.get('current_signal', '未知')}<br/>
                • 分析说明：{harmony_analysis.get('harmony_desc', '')}<br/>
                """
                
                content.append(Paragraph(harmony_text, self.normal_style))
            
            content.append(Spacer(1, 20))
            
            # 4. 突破模式识别分析
            content.append(Paragraph("四、放量突破/缩量整理模式识别", self.subtitle_style))
            
            breakout_patterns = analysis_result.get('breakout_patterns', {})
            if breakout_patterns:
                # 突破模式详情
                pattern_text = f"""
                <b>当前市场状态：</b><br/>
                • 量能状态：{breakout_patterns.get('current_state', '未知')}<br/>
                • 成交额比值：{breakout_patterns.get('current_volume_ratio', 0)}<br/>
                • 状态描述：{breakout_patterns.get('state_desc', '')}<br/>
                """
                
                content.append(Paragraph(pattern_text, self.normal_style))
                
                # 最新模式
                latest_pattern = breakout_patterns.get('latest_pattern')
                if latest_pattern:
                    latest_text = f"""
                    <b>最新识别模式：</b><br/>
                    • 模式类型：{latest_pattern.get('type', '未知')}<br/>
                    • 出现日期：{self._format_date(str(latest_pattern.get('date', '')))}<br/>
                    • 成交额比值：{latest_pattern.get('volume_ratio', 0):.2f}<br/>
                    • 价格变动：{latest_pattern.get('price_change', 0):.2f}%<br/>
                    • 信号强度：{latest_pattern.get('signal_strength', '未知')}<br/>
                    """
                    content.append(Paragraph(latest_text, self.highlight_style))
            
            content.append(Spacer(1, 30))
            
            # 5. 投资建议与风险提示
            content.append(Paragraph("五、投资建议与风险提示", self.subtitle_style))
            
            # 根据综合评分生成建议
            comprehensive_score = analysis_result.get('comprehensive_score', 50)
            
            if comprehensive_score >= 70:
                advice = """
                <b>投资建议：</b><br/>
                • 市场成交量能状态良好，可适当关注投资机会<br/>
                • 建议重点关注放量突破的优质标的<br/>
                • 保持适度仓位，注意分散投资<br/>
                <br/>
                <b>风险提示：</b><br/>
                • 注意市场波动风险，设置止损位<br/>
                • 关注宏观经济变化对市场的影响<br/>
                """
            elif comprehensive_score >= 50:
                advice = """
                <b>投资建议：</b><br/>
                • 市场成交量能状态中性，建议谨慎观察<br/>
                • 可适当配置，但需控制仓位<br/>
                • 重点关注量价配合良好的标的<br/>
                <br/>
                <b>风险提示：</b><br/>
                • 市场不确定性较大，避免重仓操作<br/>
                • 密切关注成交量变化趋势<br/>
                """
            else:
                advice = """
                <b>投资建议：</b><br/>
                • 市场成交量能状态较差，建议以观望为主<br/>
                • 控制仓位，等待更好的介入时机<br/>
                • 重点关注防御性较强的标的<br/>
                <br/>
                <b>风险提示：</b><br/>
                • 市场风险较高，需严格控制风险<br/>
                • 避免追涨杀跌，保持理性投资<br/>
                """
            
            content.append(Paragraph(advice, self.normal_style))
            
            # 添加免责声明
            content.append(Spacer(1, 20))
            disclaimer = """
            <b>免责声明：</b><br/>
            本报告基于公开市场数据进行分析，仅供参考，不构成投资建议。
            投资者应根据自身情况独立判断，投资有风险，入市需谨慎。
            """
            content.append(Paragraph(disclaimer, self.caption_style))
            
            # 生成PDF
            doc.build(content)
            
            logger.info(f"PDF报告生成成功: {pdf_path}")
            return pdf_path
            
        except Exception as e:
            logger.error(f"生成PDF报告失败: {e}")
            return ""
    
    def _format_date(self, date_str: str) -> str:
        """格式化日期显示"""
        try:
            if len(date_str) == 8:
                dt = datetime.strptime(date_str, '%Y%m%d')
                return dt.strftime('%Y年%m月%d日')
            return date_str
        except:
            return date_str
    
    def _find_chart(self, chart_files: list, chart_type: str) -> str:
        """查找指定类型的图表文件"""
        for chart_file in chart_files:
            if chart_type in chart_file:
                return chart_file
        return ""


def main():
    """主函数"""
    try:
        # 创建PDF生成器
        pdf_generator = VolumeEnergyPDFGenerator()
        
        # 生成报告（使用2025年7月1日）
        trade_date = '20250701'
        pdf_path = pdf_generator.create_comprehensive_report(trade_date)
        
        if pdf_path:
            print(f"\n成交量能分析PDF报告生成成功！")
            print(f"报告路径: {pdf_path}")
            print(f"分析日期: {trade_date}")
            
            # 尝试打开PDF文件
            try:
                import subprocess
                subprocess.Popen([pdf_path], shell=True)
                print(f"正在打开PDF报告...")
            except:
                print(f"请手动打开PDF文件查看报告")
        else:
            print("PDF报告生成失败")
        
    except Exception as e:
        logger.error(f"主函数执行失败: {e}")


if __name__ == "__main__":
    main()