金融文本信息抽取：财报、公告、新闻的三源融合解析

金融文本信息抽取：财报、公告、新闻的三源融合解析

一、个性化深度引言

一份上市公司财报 200 页，一份重大事项公告 5 页，一篇财经新闻 800 字。分析师需要在 30 分钟内看完 20 家公司的所有信息并做出判断。三个信息源，三种格式，三类噪声，却要融合成一条投资信号。

传统的做法是三个团队分别处理然后合并——信息在传递中损耗，在等待中衰减。我们尝试用一个统一的信息抽取框架来替代这个流程。见证奇迹的时刻在于：当我们将 NER 模型 + 关系抽取 + 时间轴对齐组合在一起时，财报里的数字、公告里的意图、新闻里的情绪，第一次在同一个时间线上被连接起来。

二、个性化原理剖析

三源信息融合架构

各信息源的特性与挑战

财报：结构化程度高（三表有固定格式），但 PDF 格式导致表格解析困难。关键挑战是数字抽取的精度——一个错误的小数点可能差出几亿。

公告：半结构化，关键信息在特定段落（如“重要内容提示”）。核心挑战是事件类型的分类——同一个事件可能是“利好”也可能是“利空”，取决于上下文。

新闻：完全非结构化，充斥着主观判断和情绪化表述。挑战在于区分“事实描述”和“观点评论”。

三源融合的信号加权

不同来源的信息对同一事件的描述可能矛盾。例如：财报显示利润增长 20%，但新闻标题是“XX 公司业绩暴雷”。

融合策略：

1. 事实冲突：以官方公告为准，新闻作为参考
2. 情绪冲突：保留矛盾信号，不作为过滤条件
3. 时间冲突：以最新信息为基准，旧信息标注为历史

三、个性化代码实践

from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
class SourceType(Enum):
    """信息来源类型"""
    FINANCIAL_REPORT = "财报"
    ANNOUNCEMENT = "公告"
    NEWS = "新闻"
class EventType(Enum):
    """金融事件类型"""
    MERGER = "并购重组"
    PENALTY = "监管处罚"
    DIVIDEND = "分红派息"
    SHARE_CHANGE = "增减持"
    PERFORMANCE = "业绩预告"
class Sentiment(Enum):
    """情绪标签"""
    POSITIVE = "正面"
    NEGATIVE = "负面"
    NEUTRAL = "中性"
@dataclass
class FinancialEntity:
    """金融实体"""
    name: str               # 实体名称
    entity_type: str        # 类型：公司/人物/产品
    mentions: List[str]     # 别名列表
    metadata: Dict = field(default_factory=dict)
@dataclass
class ExtractedEvent:
    """抽取的金融事件"""
    event_type: EventType
    entities: List[FinancialEntity]
    date: datetime
    source: SourceType
    confidence: float       # 置信度 0-1
    amount: Optional[float] = None  # 涉及金额
    sentiment: Sentiment = Sentiment.NEUTRAL
    raw_text: str = ""
@dataclass
class FusionResult:
    """三源融合结果"""
    company: str
    timeline: List[ExtractedEvent]  # 按时间排序的事件
    conflicts: List[Tuple[ExtractedEvent, ExtractedEvent]]  # 冲突事件对
    risk_signals: List[str]  # 风险信号
class FinancialInformationExtractor:
    """金融信息抽取引擎"""
    # 设计原因：财报核心数据关键词——三表关键项
    REPORT_FIELDS = {
        "营业收入": "revenue",
        "净利润": "net_profit",
        "扣非净利润": "adjusted_net_profit",
        "经营活动现金流": "operating_cash_flow",
        "资产负债率": "debt_ratio",
        "毛利率": "gross_margin",
        "净利率": "net_margin",
        "ROE": "roe",
    }
    # 设计原因：公告事件关键词——触发器模式
    ANNOUNCEMENT_TRIGGERS = {
        EventType.MERGER: ["收购", "合并", "重组", "资产注入"],
        EventType.PENALTY: ["处罚", "问询", "警示函", "立案调查"],
        EventType.DIVIDEND: ["分红", "派息", "送转"],
        EventType.SHARE_CHANGE: ["减持", "增持", "回购", "举牌"],
        EventType.PERFORMANCE: ["业绩预告", "业绩快报", "预增", "预减"],
    }
    def parse_financial_report(self, text: str) -> Dict[str, float]:
        """从财报文本中抽取关键财务数据"""
        # 设计原因：正则 + 规则 比端到端模型更可靠
        # 财报数字对精度要求极高，模型幻觉不可接受
        extracted = {}
        for key, field in self.REPORT_FIELDS.items():
            # 匹配模式：关键词 + 数字（含亿/万等单位）
            import re
            pattern = rf'{key}[^0-9]*?([+-]?\d+\.?\d*)\s*(亿|万|元|%|％)?'
            match = re.search(pattern, text)
            if match:
                value = float(match.group(1))
                unit = match.group(2) if match.group(2) else ""
                # 设计原因：统一转换为"亿元"单位
                if unit == "亿":
                    pass
                elif unit == "万":
                    value = value / 10000
                else:
                    value = value / 100000000  # 默认元 → 亿元
                extracted[field] = round(value, 4)
        return extracted
    def extract_events_from_announcement(
        self, text: str, date: datetime
    ) -> List[ExtractedEvent]:
        """从公告文本中抽取事件"""
        # 设计原因：基于触发词的事件检测
        # 金融公告的格式相对规范，触发词可靠性高
        events = []
        for event_type, triggers in self.ANNOUNCEMENT_TRIGGERS.items():
            for trigger in triggers:
                if trigger in text:
                    event = ExtractedEvent(
                        event_type=event_type,
                        entities=[],
                        date=date,
                        source=SourceType.ANNOUNCEMENT,
                        confidence=0.85,  # 设计原因：公告来源置信度较高
                        raw_text=text[:200],
                    )
                    events.append(event)
        return events
    def analyze_news_sentiment(self, text: str) -> Sentiment:
        """分析新闻情绪"""
        # 设计原因：金融新闻的情绪词分布有明显的领域特征
        positive_words = [
            "增长", "突破", "利好", "超预期", "创新高",
            "扩大", "提升", "改善", "回购", "增持",
        ]
        negative_words = [
            "下降", "亏损", "利空", "不及预期", "新低",
            "缩减", "下滑", "恶化", "减持", "调查",
        ]
        pos_count = sum(1 for w in positive_words if w in text)
        neg_count = sum(1 for w in negative_words if w in text)
        if pos_count > neg_count + 1:
            return Sentiment.POSITIVE
        elif neg_count > pos_count + 1:
            return Sentiment.NEGATIVE
        else:
            return Sentiment.NEUTRAL
    def fuse_sources(
        self,
        report_data: Dict[str, float],
        events: List[ExtractedEvent],
        news_sentiment: Sentiment,
        company: str,
    ) -> FusionResult:
        """三源信息融合"""
        # 设计原因：融合逻辑的三条核心规则
        timeline = sorted(events, key=lambda e: e.date)
        conflicts = []
        risk_signals = []
        # 设计原因：冲突检测——数字 vs 情绪
        # 如果财报数据向好（利润增长）但新闻情绪负面 → 风险信号
        net_profit = report_data.get("net_profit", 0)
        if net_profit > 0 and news_sentiment == Sentiment.NEGATIVE:
            risk_signals.append(
                f"矛盾信号：净利润{net_profit}亿（增长），"
                f"但新闻情绪为负面，可能有未反映在财报中的风险"
            )
        # 设计原因：处罚事件 + 减持事件 → 强风险信号
        has_penalty = any(
            e.event_type == EventType.PENALTY for e in events
        )
        has_share_decrease = any(
            e.event_type == EventType.SHARE_CHANGE and "减持" in e.raw_text
            for e in events
        )
        if has_penalty and has_share_decrease:
            risk_signals.append("风险叠加：同时存在监管处罚和股东减持事件")
        return FusionResult(
            company=company,
            timeline=timeline,
            conflicts=conflicts,
            risk_signals=risk_signals,
        )
# 使用示例
extractor = FinancialInformationExtractor()
# 模拟财报文本
report_text = """
2024年度，公司实现营业收入125.6亿元，同比增长15.2%；
归属于上市公司股东的净利润23.8亿元，同比增长18.5%；
扣除非经常性损益的净利润21.2亿元，同比增长16.8%。
"""
# 解析财报
financial_data = extractor.parse_financial_report(report_text)
print("财报数据抽取：")
for key, value in financial_data.items():
    print(f"  {key}: {value}")
# 抽取事件
announcement_text = "公司拟以自有资金回购股份，回购金额不低于1亿元。"
events = extractor.extract_events_from_announcement(
    announcement_text, datetime(2025, 1, 15)
)
print(f"\n事件抽取：{len(events)} 个事件")
# 情绪分析
news = "XX公司业绩超预期，机构上调目标价，市场反应积极。"
sentiment = extractor.analyze_news_sentiment(news)
print(f"新闻情绪：{sentiment.value}")
# 融合
result = extractor.fuse_sources(financial_data, events, sentiment, "XX公司")
print(f"\n风险信号：{result.risk_signals}")

四、个性化边界权衡

关键权衡：

1. 精度 vs 召回：财报数字抽取中，一个错误可能导致所有下游分析失效。因此优先保证精度，宁可不抽取也不抽错。
2. 可解释性 vs 性能：金融场景有强合规要求，融合决策必须可解释、可审计。端到端深度学习模型虽然性能更好，但不适合这个场景。
3. 实时性 vs 完整性：新闻情绪分析需要秒级响应（市场价格变化极快），但财报分析可以容忍分钟级延迟（信息更新频率以天计）。

五、总结

金融文本信息抽取的本质是三源异构信息的融合对齐问题。财报提供结构化的财务数据（高精度、可量化），公告提供事件触发信号（高可信度、低噪声），新闻提供市场情绪参考（高时效、主观性强）。技术方案上，建议采用规则为主、模型为辅的混合策略：财报数字用正则 + 规则（精度优先），公告事件用触发词 + 模型验证（召回优先），新闻情绪用 FinBERT 等专用模型。融合层的核心逻辑是冲突检测——当不同来源的信号矛盾时，以官方公告为准，保留矛盾信号作为风险提示。最终输出的结构化数据是量化模型的输入，而非直接的投资结论。