hahaju/aiagents-stock
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
oficcejo
2026-02-27 15:15:49 +08:00
parent cfda09f3e4
commit 63628ffdd5
7 changed files with 2014 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
macro_cycle_pdf.py
+427
View File
@@ -0,0 +1,427 @@
"""
宏观周期分析 - PDF报告生成模块
生成康波周期 × 美林投资时钟 × 中国政策分析的完整PDF报告
"""
from reportlab.lib import colors
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, PageBreak
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
from reportlab.lib.enums import TA_CENTER, TA_LEFT, TA_JUSTIFY
from datetime import datetime
import os
import tempfile
import re
class MacroCyclePDFGenerator:
"""宏观周期分析PDF报告生成器"""
def __init__(self):
"""初始化PDF生成器"""
self.setup_fonts()
def setup_fonts(self):
"""设置中文字体"""
try:
font_paths = [
'C:/Windows/Fonts/msyh.ttc', # 微软雅黑
'C:/Windows/Fonts/simsun.ttc', # 宋体
'C:/Windows/Fonts/simhei.ttf', # 黑体
'/usr/share/fonts/truetype/wqy/wqy-microhei.ttc', # Linux
'/usr/share/fonts/truetype/droid/DroidSansFallbackFull.ttf',
]
for font_path in font_paths:
if os.path.exists(font_path):
try:
pdfmetrics.registerFont(TTFont('ChineseFont', font_path))
self.chinese_font = 'ChineseFont'
print(f"[宏观PDF] 成功加载字体: {font_path}")
return
except:
continue
self.chinese_font = 'Helvetica'
print("[宏观PDF] 警告: 未找到中文字体，使用默认字体")
except Exception as e:
print(f"[宏观PDF] 字体设置失败: {e}")
self.chinese_font = 'Helvetica'
def generate_pdf(self, result_data: dict, output_path: str = None) -> str:
"""
生成宏观周期分析PDF报告
Args:
result_data: 分析结果数据
output_path: 输出路径，如果为None则生成临时文件
Returns:
PDF文件路径
"""
try:
if output_path is None:
temp_dir = tempfile.gettempdir()
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_path = os.path.join(temp_dir, f"宏观周期报告_{timestamp}.pdf")
doc = SimpleDocTemplate(
output_path,
pagesize=A4,
rightMargin=0.5*inch,
leftMargin=0.5*inch,
topMargin=0.5*inch,
bottomMargin=0.5*inch
)
story = []
# 1. 标题页
story.extend(self._create_title_page(result_data))
story.append(PageBreak())
# 2. 首席宏观策略师综合研判（最重要，放最前面）
story.extend(self._create_chief_section(result_data))
story.append(PageBreak())
# 3. 康波周期分析
story.extend(self._create_kondratieff_section(result_data))
story.append(PageBreak())
# 4. 美林投资时钟分析
story.extend(self._create_merrill_section(result_data))
story.append(PageBreak())
# 5. 中国政策分析
story.extend(self._create_policy_section(result_data))
# 6. 结束语
story.extend(self._create_ending())
# 生成PDF
doc.build(story)
print(f"[宏观PDF] 报告生成成功: {output_path}")
return output_path
except Exception as e:
print(f"[宏观PDF] 生成失败: {e}")
import traceback
traceback.print_exc()
raise
def _clean_markdown(self, text: str) -> str:
"""清理Markdown标记，转换为适合PDF的纯文本/HTML"""
if not text:
return ""
# 移除markdown粗体 **text** → text
text = re.sub(r'\*\*(.*?)\*\*', r'<b>\1</b>', text)
# 移除markdown斜体 *text* → text
text = re.sub(r'\*(.*?)\*', r'<i>\1</i>', text)
# 移除markdown标题 ## → 空
text = re.sub(r'^#{1,6}\s+', '', text, flags=re.MULTILINE)
# 移除markdown链接 [text](url) → text
text = re.sub(r'\[(.*?)\]\(.*?\)', r'\1', text)
# 移除markdown表格分隔线
text = re.sub(r'\|[-:]+\|[-:| ]+\|', '', text)
# 替换换行
text = text.replace('\n', '<br/>')
return text
def _split_text_to_paragraphs(self, text: str, styles: dict, max_chars: int = 0) -> list:
"""将长文本分段为多个Paragraph，避免单段过长溢出"""
elements = []
if not text:
return elements
if max_chars > 0 and len(text) > max_chars:
text = text[:max_chars] + "...（更多内容请查看网页版完整报告）"
# 按段落分割
paragraphs = text.split('\n\n')
for para in paragraphs:
para = para.strip()
if not para:
continue
# 清理markdown
cleaned = self._clean_markdown(para)
if cleaned:
try:
elements.append(Paragraph(cleaned, styles['Small']))
elements.append(Spacer(1, 0.08*inch))
except Exception:
# 如果格式化失败，用纯文本
plain = re.sub(r'<[^>]+>', '', cleaned)
elements.append(Paragraph(plain, styles['Small']))
elements.append(Spacer(1, 0.08*inch))
return elements
def _create_title_page(self, data: dict) -> list:
"""创建标题页"""
styles = self._get_styles()
elements = []
elements.append(Spacer(1, 1.5*inch))
# 主标题
elements.append(Paragraph("宏观周期分析报告", styles['Title']))
elements.append(Spacer(1, 0.3*inch))
# 副标题
elements.append(Paragraph(
"康波周期 × 美林投资时钟 × 中国政策分析",
styles['Heading2']
))
elements.append(Spacer(1, 0.8*inch))
# 报告信息
timestamp = data.get('timestamp', datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
info_text = f"""
<para align=center>
<b>生成时间:</b> {timestamp}<br/>
<b>分析框架:</b> 三维周期定位系统<br/>
<b>AI分析师:</b> 4位专业分析师协同研判<br/>
<b>分析维度:</b> 康波长周期 · 美林中短周期 · 中国政策环境<br/>
<b>数据来源:</b> AKShare宏观经济数据
</para>
"""
elements.append(Paragraph(info_text, styles['Normal']))
elements.append(Spacer(1, 0.5*inch))
# 分析师团队
team_text = """
<para align=center>
<b>AI分析师团队:</b><br/>
康波周期分析师 · 美林时钟分析师 · 中国政策分析师 · 首席宏观策略师
</para>
"""
elements.append(Paragraph(team_text, styles['Normal']))
elements.append(Spacer(1, 0.8*inch))
# 免责声明
elements.append(Paragraph(
"<para align=center><i>本报告由AI系统自动生成，仅供学习研究参考，不构成投资建议。<br/>"
"周期理论是认知框架而非精确预测工具。投资有风险，入市需谨慎。</i></para>",
styles['Small']
))
return elements
def _create_chief_section(self, data: dict) -> list:
"""创建首席宏观策略师综合研判部分"""
styles = self._get_styles()
elements = []
elements.append(Paragraph("一、首席宏观策略师 — 综合研判", styles['Heading1']))
elements.append(Spacer(1, 0.15*inch))
elements.append(Paragraph(
"<i>整合康波周期、美林投资时钟、中国政策三个维度，构建周期仪表盘，给出最终综合投资策略。</i>",
styles['Small']
))
elements.append(Spacer(1, 0.15*inch))
chief = data.get('agents_analysis', {}).get('chief', {})
analysis = chief.get('analysis', '暂无分析结果')
elements.extend(self._split_text_to_paragraphs(analysis, styles, max_chars=5000))
return elements
def _create_kondratieff_section(self, data: dict) -> list:
"""创建康波周期分析部分"""
styles = self._get_styles()
elements = []
elements.append(Paragraph("二、康波周期分析 — 60年长周期定位", styles['Heading1']))
elements.append(Spacer(1, 0.15*inch))
elements.append(Paragraph(
"<i>基于康德拉季耶夫长波理论（周金涛\"人生发财靠康波\"），判断当前处于第五轮信息技术康波的阶段位置。</i>",
styles['Small']
))
elements.append(Spacer(1, 0.15*inch))
kondratieff = data.get('agents_analysis', {}).get('kondratieff', {})
analysis = kondratieff.get('analysis', '暂无分析结果')
elements.extend(self._split_text_to_paragraphs(analysis, styles, max_chars=5000))
return elements
def _create_merrill_section(self, data: dict) -> list:
"""创建美林投资时钟分析部分"""
styles = self._get_styles()
elements = []
elements.append(Paragraph("三、美林投资时钟 — 中短周期定位", styles['Heading1']))
elements.append(Spacer(1, 0.15*inch))
elements.append(Paragraph(
"<i>基于经济增长与通胀两大维度，结合中国政策方向（第三维度），判断当前处于美林时钟的哪个象限。</i>",
styles['Small']
))
elements.append(Spacer(1, 0.15*inch))
merrill = data.get('agents_analysis', {}).get('merrill', {})
analysis = merrill.get('analysis', '暂无分析结果')
elements.extend(self._split_text_to_paragraphs(analysis, styles, max_chars=5000))
return elements
def _create_policy_section(self, data: dict) -> list:
"""创建中国政策分析部分"""
styles = self._get_styles()
elements = []
elements.append(Paragraph("四、中国政策环境分析", styles['Heading1']))
elements.append(Spacer(1, 0.15*inch))
elements.append(Paragraph(
"<i>深度分析货币政策、财政政策、产业政策、房地产政策，评估政策对周期的影响和投资机会。</i>",
styles['Small']
))
elements.append(Spacer(1, 0.15*inch))
policy = data.get('agents_analysis', {}).get('policy', {})
analysis = policy.get('analysis', '暂无分析结果')
elements.extend(self._split_text_to_paragraphs(analysis, styles, max_chars=5000))
return elements
def _create_ending(self) -> list:
"""创建结束语"""
styles = self._get_styles()
elements = []
elements.append(Spacer(1, 0.5*inch))
elements.append(Paragraph(
"<para align=center><i>--- 报告结束 ---<br/>"
"本报告由宏观周期AI分析系统自动生成<br/>"
"康波是罗盘，美林是航海图，政策是季风<br/>"
"愿你在经济的海洋中，驶向属于自己的财富彼岸</i></para>",
styles['Normal']
))
return elements
def _get_styles(self) -> dict:
"""获取样式"""
styles = getSampleStyleSheet()
custom_styles = {
'Title': ParagraphStyle(
'MacroTitle',
parent=styles['Title'],
fontName=self.chinese_font,
fontSize=26,
textColor=colors.HexColor('#302b63'),
spaceAfter=30,
alignment=TA_CENTER
),
'Heading1': ParagraphStyle(
'MacroHeading1',
parent=styles['Heading1'],
fontName=self.chinese_font,
fontSize=16,
textColor=colors.HexColor('#0f0c29'),
spaceAfter=12,
spaceBefore=12
),
'Heading2': ParagraphStyle(
'MacroHeading2',
parent=styles['Heading2'],
fontName=self.chinese_font,
fontSize=14,
textColor=colors.HexColor('#302b63'),
spaceAfter=10,
spaceBefore=10,
alignment=TA_CENTER
),
'Normal': ParagraphStyle(
'MacroNormal',
parent=styles['Normal'],
fontName=self.chinese_font,
fontSize=11,
leading=16,
alignment=TA_JUSTIFY
),
'Small': ParagraphStyle(
'MacroSmall',
parent=styles['Normal'],
fontName=self.chinese_font,
fontSize=9,
leading=14,
alignment=TA_LEFT
)
}
return custom_styles
def generate_macro_cycle_markdown(result_data: dict) -> str:
"""生成宏观周期分析的Markdown报告"""
parts = []
timestamp = result_data.get('timestamp', datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
parts.append("# 🧭 宏观周期分析报告\n")
parts.append(f"**生成时间**: {timestamp}\n")
parts.append("**分析框架**: 康波周期 × 美林投资时钟 × 中国政策分析\n")
parts.append("---\n")
agents = result_data.get('agents_analysis', {})
# 首席策略师
chief = agents.get('chief', {})
if chief:
parts.append("## 👔 一、首席宏观策略师 — 综合研判\n")
parts.append(chief.get('analysis', '暂无分析结果'))
parts.append("\n\n---\n")
# 康波周期
kondratieff = agents.get('kondratieff', {})
if kondratieff:
parts.append("## 🌊 二、康波周期分析 — 60年长周期定位\n")
parts.append(kondratieff.get('analysis', '暂无分析结果'))
parts.append("\n\n---\n")
# 美林时钟
merrill = agents.get('merrill', {})
if merrill:
parts.append("## ⏰ 三、美林投资时钟 — 中短周期定位\n")
parts.append(merrill.get('analysis', '暂无分析结果'))
parts.append("\n\n---\n")
# 政策分析
policy = agents.get('policy', {})
if policy:
parts.append("## 🏛️ 四、中国政策环境分析\n")
parts.append(policy.get('analysis', '暂无分析结果'))
parts.append("\n\n---\n")
# 免责声明
parts.append("\n> ⚠️ **免责声明**: 本报告由AI系统自动生成，仅供学习研究参考，不构成投资建议。")
parts.append("周期理论是认知框架而非精确预测工具。投资有风险，入市需谨慎。\n")
return "\n".join(parts)
# 测试
if __name__ == "__main__":
print("=" * 60)
print("测试宏观周期PDF生成器")
print("=" * 60)
test_data = {
"success": True,
"timestamp": "2026-02-27 14:00:00",
"agents_analysis": {
"chief": {"analysis": "综合研判测试内容..."},
"kondratieff": {"analysis": "康波分析测试内容..."},
"merrill": {"analysis": "美林时钟测试内容..."},
"policy": {"analysis": "政策分析测试内容..."},
}
}
generator = MacroCyclePDFGenerator()
output_path = generator.generate_pdf(test_data)
print(f"测试PDF生成: {output_path}")