GLM-4-9B-Chat-1M实战：财务报表自动分析与可视化

路怜涯

322人浏览 · 2026-02-14 00:52:26

路怜涯 · 2026-02-14 00:52:26 发布

GLM-4-9B-Chat-1M实战：财务报表自动分析与可视化

1. 引言：财务分析的智能化革命

财务报表分析是企业和投资者了解公司经营状况的核心手段，但传统的人工分析方式面临诸多挑战。一份完整的上市公司年报往往包含上百页的财务数据、附注说明和经营讨论，分析师需要花费数小时甚至数天时间才能完成深度分析。

想象一下这样的场景：你收到一家公司的年度财报PDF文档，足足有150页。你需要手动提取关键财务数据、计算各种比率指标、识别异常变动趋势，最后还要生成可视化的分析报告。这个过程不仅耗时耗力，还容易因为人为因素出现遗漏或错误。

GLM-4-9B-Chat-1M模型的问世，为财务分析带来了全新的解决方案。这个拥有100万tokens超长上下文处理能力的模型，可以一次性吞下整份财报文档，像一位不知疲倦的财务专家一样，快速完成数据提取、分析和可视化工作。

本文将带你一步步实现财务报表的自动分析与可视化，让你体验AI驱动的智能财务分析全流程。

2. 环境准备与快速部署

2.1 硬件要求与系统配置

首先确保你的设备满足以下最低要求：

GPU显存：至少8GB（用于4-bit量化版本）
系统内存：32GB以上
存储空间：20GB可用空间
操作系统：Linux推荐，Windows/macOS也可运行

2.2 一键部署GLM-4-9B-Chat-1M

通过Docker快速部署环境：

# 拉取预配置的镜像
docker pull csdnmirror/glm-4-9b-chat-1m

# 运行容器
docker run -p 8080:8080 --gpus all -v $(pwd)/data:/app/data csdnmirror/glm-4-9b-chat-1m

或者使用Python环境直接安装：

# 创建虚拟环境
conda create -n financial-ai python=3.10
conda activate financial-ai

# 安装核心依赖
pip install transformers torch pandas matplotlib seaborn streamlit
pip install pdfplumber python-docx  # 文档处理库

2.3 验证模型运行

简单的测试脚本确保模型正常工作：

from transformers import AutoModelForCausalLM, AutoTokenizer

model_path = "THUDM/glm-4-9b-chat-1m"
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    trust_remote_code=True
)

# 测试短文本处理
test_query = "请简单介绍你自己"
inputs = tokenizer.apply_chat_template([{"role": "user", "content": test_query}], 
                                      add_generation_prompt=True, 
                                      return_tensors="pt").to(model.device)

outputs = model.generate(inputs, max_length=200)
print(tokenizer.decode(outputs[0], skip_special_tokens=True))

3. 财务报表处理与分析流程

3.1 财务文档解析与数据提取

上市公司财报通常有PDF和Word两种格式，我们需要先将其转换为纯文本：

import pdfplumber
from docx import Document

def extract_text_from_pdf(pdf_path):
    """从PDF提取文本内容"""
    full_text = ""
    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            text = page.extract_text()
            if text:
                full_text += text + "\n"
    return full_text

def extract_text_from_docx(docx_path):
    """从Word文档提取文本"""
    doc = Document(docx_path)
    return "\n".join([paragraph.text for paragraph in doc.paragraphs])

def preprocess_financial_text(text):
    """预处理财务文本，增强关键信息"""
    # 识别并标记关键财务部分
    sections = {
        "balance_sheet": ["资产负债表", "财务状况表"],
        "income_statement": ["利润表", "损益表"],
        "cash_flow": ["现金流量表"],
        "financial_ratios": ["财务比率", "关键指标"]
    }
    
    processed_text = text
    # 这里可以添加更多的文本清洗和增强逻辑
    return processed_text

3.2 构建财务分析提示词模板

有效的提示词是获得高质量分析结果的关键：

def create_financial_analysis_prompt(financial_text, analysis_type="comprehensive"):
    """创建财务分析提示词"""
    
    base_prompt = """你是一位资深财务分析师，请对以下财务报表进行专业分析：

{financial_text}

请完成以下分析任务：
1. 提取关键财务数据（营业收入、净利润、资产负债等）
2. 计算重要财务比率（毛利率、净利率、资产负债率等）
3. 识别财务趋势和异常点
4. 评估公司财务健康状况
5. 提供投资建议分析

请以结构化JSON格式返回分析结果，包含以下字段：
- financial_data: 关键财务数据
- financial_ratios: 财务比率计算
- trend_analysis: 趋势分析
- risk_assessment: 风险评估
- investment_advice: 投资建议
"""
    
    if analysis_type == "quick":
        base_prompt += "\n请提供简洁版分析，重点关注关键指标和风险点。"
    elif analysis_type == "detailed":
        base_prompt += "\n请提供详细分析，包括同比环比数据对比和行业对比。"
    
    return base_prompt.format(financial_text=financial_text[:500000])  # 控制输入长度

4. 实现自动化财务分析系统

4.1 核心分析函数实现

import json
import torch
from typing import Dict, Any

def analyze_financial_statement(model, tokenizer, financial_text, max_length=10000):
    """执行财务分析"""
    
    prompt = create_financial_analysis_prompt(financial_text)
    
    # 准备模型输入
    inputs = tokenizer.apply_chat_template(
        [{"role": "user", "content": prompt}],
        add_generation_prompt=True,
        return_tensors="pt"
    ).to(model.device)
    
    # 生成分析结果
    with torch.no_grad():
        outputs = model.generate(
            inputs,
            max_length=max_length,
            temperature=0.1,  # 低温度确保输出稳定性
            do_sample=True,
            top_p=0.9,
            pad_token_id=tokenizer.eos_token_id
        )
    
    # 提取生成文本
    response = outputs[0][inputs.shape[1]:]
    analysis_text = tokenizer.decode(response, skip_special_tokens=True)
    
    # 尝试解析JSON结果
    try:
        # 提取JSON部分
        json_start = analysis_text.find('{')
        json_end = analysis_text.rfind('}') + 1
        json_str = analysis_text[json_start:json_end]
        analysis_data = json.loads(json_str)
    except json.JSONDecodeError:
        # 如果JSON解析失败，返回原始文本
        analysis_data = {"raw_analysis": analysis_text}
    
    return analysis_data

def batch_analyze_financials(model, tokenizer, financial_texts):
    """批量分析多个财务文档"""
    results = []
    for text in financial_texts:
        try:
            analysis = analyze_financial_statement(model, tokenizer, text)
            results.append(analysis)
        except Exception as e:
            print(f"分析过程中出错: {e}")
            results.append({"error": str(e)})
    return results

4.2 财务可视化生成

基于分析结果生成可视化图表：

import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
from io import BytesIO
import base64

def create_financial_visualizations(analysis_data, company_name=""):
    """创建财务可视化图表"""
    
    visualizations = {}
    
    # 确保有数据可可视化
    if 'financial_data' in analysis_data and analysis_data['financial_data']:
        financial_data = analysis_data['financial_data']
        
        # 创建财务指标趋势图
        if 'revenue' in financial_data and 'net_profit' in financial_data:
            fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8))
            
            # 营业收入趋势
            revenue_data = financial_data['revenue']
            if isinstance(revenue_data, dict) and 'history' in revenue_data:
                years = list(revenue_data['history'].keys())
                values = list(revenue_data['history'].values())
                ax1.plot(years, values, marker='o', linewidth=2)
                ax1.set_title(f'{company_name}营业收入趋势' if company_name else '营业收入趋势')
                ax1.set_ylabel('金额（万元）')
                ax1.grid(True, alpha=0.3)
            
            # 净利润趋势
            profit_data = financial_data['net_profit']
            if isinstance(profit_data, dict) and 'history' in profit_data:
                years = list(profit_data['history'].keys())
                values = list(profit_data['history'].values())
                ax2.plot(years, values, marker='s', color='orange', linewidth=2)
                ax2.set_title(f'{company_name}净利润趋势' if company_name else '净利润趋势')
                ax2.set_ylabel('金额（万元）')
                ax2.grid(True, alpha=0.3)
            
            plt.tight_layout()
            visualizations['trend_analysis'] = fig_to_base64(fig)
            plt.close(fig)
    
    # 财务比率雷达图
    if 'financial_ratios' in analysis_data:
        ratios = analysis_data['financial_ratios']
        if ratios:
            categories = ['盈利能力', '偿债能力', '运营能力', '成长能力']
            values = [
                ratios.get('profit_margin', 0),
                ratios.get('current_ratio', 0),
                ratios.get('asset_turnover', 0),
                ratios.get('revenue_growth', 0)
            ]
            
            fig = plt.figure(figsize=(8, 8))
            ax = fig.add_subplot(111, polar=True)
            
            angles = [n / float(len(categories)) * 2 * 3.14159 for n in range(len(categories))]
            angles += angles[:1]  # 闭合雷达图
            
            values += values[:1]
            ax.plot(angles, values, linewidth=2)
            ax.fill(angles, values, alpha=0.25)
            
            ax.set_xticks(angles[:-1])
            ax.set_xticklabels(categories)
            ax.set_title('财务能力雷达图')
            
            visualizations['radar_chart'] = fig_to_base64(fig)
            plt.close(fig)
    
    return visualizations

def fig_to_base64(fig):
    """将matplotlib图表转换为base64字符串"""
    buf = BytesIO()
    fig.savefig(buf, format='png', dpi=100, bbox_inches='tight')
    buf.seek(0)
    return base64.b64encode(buf.read()).decode('utf-8')

5. 构建完整的财务分析应用

5.1 Streamlit Web应用集成

创建一个用户友好的Web界面：

import streamlit as st
import json

def main():
    st.title(" 智能财务报表分析系统")
    st.write("基于GLM-4-9B-Chat-1M的智能财务分析工具")
    
    # 文件上传区域
    uploaded_file = st.file_uploader("上传财务报表文件", type=['pdf', 'docx', 'txt'])
    
    analysis_type = st.radio("分析深度", ["快速分析", "详细分析"], horizontal=True)
    
    if st.button("开始分析") and uploaded_file:
        with st.spinner("正在处理和分析文档..."):
            # 读取文件内容
            if uploaded_file.type == "application/pdf":
                text_content = extract_text_from_pdf(uploaded_file)
            elif uploaded_file.type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
                text_content = extract_text_from_docx(uploaded_file)
            else:
                text_content = str(uploaded_file.read(), 'utf-8')
            
            # 执行分析
            analysis_result = analyze_financial_statement(model, tokenizer, text_content)
            
            # 显示分析结果
            st.subheader(" 财务分析结果")
            
            if 'financial_data' in analysis_result:
                st.json(analysis_result['financial_data'])
            
            if 'financial_ratios' in analysis_result:
                col1, col2, col3 = st.columns(3)
                ratios = analysis_result['financial_ratios']
                
                with col1:
                    st.metric("毛利率", f"{ratios.get('gross_margin', 0):.2f}%")
                    st.metric("净利率", f"{ratios.get('net_margin', 0):.2f}%")
                
                with col2:
                    st.metric("资产负债率", f"{ratios.get('debt_ratio', 0):.2f}%")
                    st.metric("流动比率", f"{ratios.get('current_ratio', 0):.2f}")
                
                with col3:
                    st.metric("ROE", f"{ratios.get('roe', 0):.2f}%")
                    st.metric("营收增长率", f"{ratios.get('revenue_growth', 0):.2f}%")
            
            # 显示可视化图表
            if 'financial_data' in analysis_result:
                visuals = create_financial_visualizations(analysis_result, 
                                                         uploaded_file.name.split('.')[0])
                
                for viz_name, viz_data in visuals.items():
                    st.image(f"data:image/png;base64,{viz_data}", 
                            caption=viz_name.replace('_', ' ').title())
            
            # 显示投资建议
            if 'investment_advice' in analysis_result:
                st.subheader(" 投资建议")
                st.write(analysis_result['investment_advice'])
    
    # 添加示例文件下载
    with st.expander("需要示例文件？"):
        st.write("下载示例财务报表进行测试")
        # 这里可以添加示例文件下载链接

if __name__ == "__main__":
    # 初始化模型（在实际应用中应该缓存模型加载）
    @st.cache_resource
    def load_model():
        tokenizer = AutoTokenizer.from_pretrained("THUDM/glm-4-9b-chat-1m", 
                                                trust_remote_code=True)
        model = AutoModelForCausalLM.from_pretrained(
            "THUDM/glm-4-9b-chat-1m",
            torch_dtype=torch.bfloat16,
            device_map="auto",
            trust_remote_code=True
        )
        return model, tokenizer
    
    model, tokenizer = load_model()
    main()

5.2 高级功能扩展

def comparative_analysis(model, tokenizer, company_texts, company_names):
    """比较多家公司财务表现"""
    
    comparative_prompt = f"""请对比分析以下几家公司的财务状况：

公司列表：{', '.join(company_names)}

请从以下维度进行对比分析：
1. 盈利能力对比（毛利率、净利率、ROE）
2. 财务安全性对比（资产负债率、流动比率）
3. 成长性对比（营收增长率、利润增长率）
4. 运营效率对比（资产周转率）
5. 综合排名和建议

请以表格形式呈现对比结果，并给出投资建议。"""
    
    inputs = tokenizer.apply_chat_template(
        [{"role": "user", "content": comparative_prompt}],
        add_generation_prompt=True,
        return_tensors="pt"
    ).to(model.device)
    
    outputs = model.generate(inputs, max_length=8000)
    analysis = tokenizer.decode(outputs[0][inputs.shape[1]:], skip_special_tokens=True)
    
    return analysis

def financial_early_warning(analysis_data, industry_benchmarks):
    """财务风险早期预警"""
    warnings = []
    
    ratios = analysis_data.get('financial_ratios', {})
    financials = analysis_data.get('financial_data', {})
    
    # 偿债能力预警
    if ratios.get('current_ratio', 2) < 1.2:
        warnings.append(" 流动比率低于警戒线，短期偿债能力不足")
    
    if ratios.get('debt_ratio', 0) > 70:
        warnings.append(" 资产负债率过高，财务风险较大")
    
    # 盈利能力预警
    if ratios.get('net_margin', 0) < industry_benchmarks.get('net_margin', 5):
        warnings.append(" 净利率低于行业平均水平")
    
    # 成长性预警
    if financials.get('revenue_growth', 0) < 0:
        warnings.append(" 营业收入出现负增长")
    
    return warnings