LLM在财务数据提取与计算中的实践应用

1. 项目概述：LLM在财务数据提取与计算中的应用实践

在金融分析领域，快速准确地从企业年报中提取关键财务指标是分析师的基础工作。传统方法依赖人工查找或简单的文本匹配，效率低下且容易出错。最近我在处理苹果公司2023财年10-K年报（约90页，7万token）时，尝试利用大语言模型(LLM)实现自动化数值提取和计算，效果显著。这个案例展示了LLM处理结构化财务数据的潜力，特别在以下场景：

• 直接从非结构化文本中定位特定数值（如总营收、研发费用）
• 执行基于提取值的数学运算（如百分比计算）
• 生成机器可读的结构化输出（JSON格式）

测试中对比了两种技术方案：当文档长度在模型上下文窗口内（如128K token模型）时，单次传入可获得精确结果；对于超长文档，采用分块+向量检索的RAG方法也能保持较高准确率。本文将详细解析实现过程，分享调优技巧，并探讨提升数值可靠性的实用方法。

2. 环境准备与数据获取

2.1 开发环境配置

选择Qwen3.5系列模型作为基础，因其对长上下文和数值处理表现良好。为避免网络问题，预先配置国内镜像源：

python复制import os
os.environ['HF_ENDPOINT'] = "https://hf-mirror.com"
os.environ['OPENAI_API_KEY'] = "sk-xxxxxx"  # 替换为实际API密钥
os.environ['OPENAI_BASE_URL'] = "https://llm_provider.com/v1"
model_name = "qwen3-xxxx"  # 实际模型名称

提示：如果处理中文财报，建议选择在中文金融语料上微调过的模型，如Qwen-Finance或ChatGLM-Finance，它们对财务术语的理解更准确。

2.2 数据源处理

从SEC EDGAR下载苹果公司2023年10-K年报（HTML格式），保存为本地文本文件：

1. 访问SEC EDGAR
2. 全选网页内容复制到文本编辑器
3. 保存为aapl-20230930.txt

使用tiktoken库计算文档token量，判断是否需要分块处理：

python复制import tiktoken

def num_tokens_from_string(text: str, encoding_name: str = "cl100k_base") -> int:
    encoding = tiktoken.get_encoding(encoding_name)
    return len(encoding.encode(text))

with open("./aapl-20230930.txt") as f:
    full_text = f.read()

tokens = num_tokens_from_string(full_text)
print(f"文档token数: {tokens}")  # 示例输出: 45185

当token数超过模型上下文窗口（如128K）的80%时，建议启用分块策略，保留部分空间给提示词和输出。

3. 单次传入完整文档的提取方法

3.1 提示词工程设计

核心思路是通过结构化提示引导模型精确输出。关键要素包括：

• 明确角色设定（财务专家）
• 指定输出格式（JSON）
• 限定数值单位（百万美元）
• 降低temperature减少随机性

python复制from openai import OpenAI
client = OpenAI()

def ask_model(document, questions):
    prompt = f"""你是一个财务分析专家。以下是苹果公司2023财年10-K年报的部分文本。
    
请根据文本回答以下问题，并以JSON格式返回结果。JSON键为问题编号，值为对应的答案（数值或字符串）。

文本内容：
{document}

问题：
{questions}
请直接输出JSON，不要包含其他文字。"""
    response = client.chat.completions.create(
        model=model_name,
        messages=[{"role": "user", "content": prompt}],
        temperature=0,  # 确保数值稳定
        max_tokens=500
    )
    return response.choices[0].message.content

3.2 查询执行与结果验证

定义典型财务问题并调用API：

python复制question_list = [
    "1. 2023财年总营收（Total net sales）是多少？请以百万美元为单位，只输出数字。",
    "2. 2023财年研发费用（Research and Development）是多少？请以百万美元为单位，只输出数字。",
    "3. 研发费用占总营收的比例是多少？请以百分比形式输出，保留两位小数。"
]
questions = "\n".join(question_list)
result = ask_model(full_text, questions)

结果验证方法：

python复制import json
import re

ground_truth = {
    "1": 383285,  # 单位：百万美元
    "2": 29915,
    "3": 7.804     # 29915/383285≈7.804%
}

def validate_results(llm_output):
    try:
        data = json.loads(re.search(r'\{.*\}', llm_output, re.DOTALL).group())
        for q, pred in data.items():
            gt = ground_truth[q]
            if q == "3":
                pred_val = float(pred.strip('%'))
                print(f"Q{q}: 预测 {pred_val}% vs 真实 {gt}%，误差 {abs(pred_val-gt):.2f}%")
            else:
                print(f"Q{q}: 预测 {pred} vs 真实 {gt}，误差 {abs(int(pred)-gt)}")
    except Exception as e:
        print("验证失败:", e)

validate_results(result)

注意事项：实际应用中建议添加循环重试机制，当误差超过阈值（如1%）时自动重新查询，避免偶发错误影响下游分析。

4. 分块检索增强生成(RAG)方案

4.1 文档分块与向量化

对于超长文档，采用滑动窗口分块策略：

python复制from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

def chunk_text(text, chunk_size=1000, overlap=100):
    words = text.split()
    return [' '.join(words[i:i+chunk_size]) 
           for i in range(0, len(words), chunk_size - overlap)]

chunks = chunk_text(full_text)
print(f"生成{len(chunks)}个文本块")  # 示例：32个

# 构建向量索引
model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = model.encode(chunks)
index = faiss.IndexFlatL2(embeddings.shape[1])
index.add(np.array(embeddings))

4.2 检索增强问答流程

实现基于语义检索的问答：

python复制def retrieve_chunks(query, k=3):
    query_emb = model.encode([query])
    distances, indices = index.search(query_emb, k)
    return [chunks[i] for i in indices[0]]

def rag_qa(questions):
    combined_query = " ".join(questions.values())
    context = "\n\n---\n\n".join(retrieve_chunks(combined_query, k=5))
    
    prompt = f"""根据以下从苹果10-K年报中提取的相关段落，回答问题。以JSON格式返回。

相关段落：
{context}

问题：
1. 2023财年总营收（Total net sales）是多少？以百万美元为单位。
2. 2023财年研发费用（Research and Development）是多少？以百万美元为单位。
3. 研发费用占总营收的比例是多少？以百分比形式。

要求：只输出JSON，如：
{{"1": 383285, "2": 29915, "3": "7.81%"}}"""
    
    response = client.chat.completions.create(
        model=model_name,
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        max_tokens=300
    )
    return response.choices[0].message.content

questions_dict = {
    "1": "2023财年总营收",
    "2": "2023财年研发费用",
    "3": "研发费用占总营收的比例"
}
rag_result = rag_qa(questions_dict)
validate_results(rag_result)

实操心得：调整检索块数量(k值)时需权衡——k太小可能遗漏关键信息，k太大会增加无关噪声。建议通过少量测试确定最佳k值（通常3-5之间）。

5. 数值可靠性提升技巧

5.1 思维链(CoT)提示

要求模型展示推理过程：

python复制cot_prompt = """
请逐步回答：
1. 在文本中定位总营收数据的位置（引用附近句子）
2. 同样定位研发费用数据
3. 展示比例计算过程：(研发费用/总营收)*100
最后以JSON输出最终结果。
"""

5.2 多采样一致性验证

通过多次采样取众数或平均值：

python复制def multi_sample_consensus(prompt, n=3):
    responses = []
    for _ in range(n):
        response = client.chat.completions.create(
            model=model_name,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,  # 适当增加多样性
            max_tokens=500
        )
        responses.append(response.choices[0].message.content)
    
    # 提取数值并统计众数
    values = [extract_values(r) for r in responses]
    from statistics import mode
    return mode(values)

5.3 外部计算工具集成

让模型输出计算表达式，由Python执行：

python复制calc_prompt = """
请按以下格式输出：
{
  "expression": "(研发费用/总营收)*100",
  "values": {"总营收": 383285, "研发费用": 29915}
}
"""

def safe_calc(llm_output):
    data = json.loads(llm_output)
    try:
        result = eval(data["expression"], {}, data["values"])
        return {"3": f"{result:.2f}%"}
    except:
        return {"error": "计算失败"}

避坑指南：避免直接eval不可信输入。生产环境应使用ast.literal_eval或预定义安全计算函数。

6. 性能优化与扩展应用

6.1 处理速度优化技巧

• 并行化处理：对多个独立查询使用异步请求

python复制import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def async_query(prompt):
    response = await async_client.chat.completions.create(
        model=model_name,
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )
    return response.choices[0].message.content

• 缓存机制：对相同文本块缓存向量嵌入结果

python复制from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_embed(text):
    return model.encode(text)

6.2 扩展到其他财务场景

• 利润表分析：自动提取毛利率、运营利润率等指标
• 现金流分析：识别经营/投资/筹资活动现金流
• 风险指标计算：如负债率、流动比率等

python复制advanced_questions = [
    "计算营运利润率：(Operating Income/Net Sales)*100",
    "列出前三大营收构成板块及占比",
    "计算当前资产负债率：Total Liabilities/Total Assets"
]

6.3 结果可视化集成

将输出JSON自动转换为图表：

python复制import matplotlib.pyplot as plt

def plot_finance_data(data):
    labels = ['总营收', '研发费用']
    values = [data["1"], data["2"]]
    
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(10,4))
    ax1.bar(labels, values)
    ax1.set_title('绝对数值（百万美元）')
    
    ax2.pie([data["3"], 100-float(data["3"].strip('%'))], 
           labels=['研发占比', '其他'], autopct='%1.1f%%')
    ax2.set_title('研发费用占比')
    plt.show()

在实际项目中，这种LLM驱动的财务数据分析方法相比传统手动处理效率提升显著。一个典型的使用场景是季度财报发布后的快速分析——原本需要分析师数小时的工作，现在通过自动化流程可在几分钟内完成初步数据提取和基础计算，让人类专家更专注于高阶分析。