LLM alpha 与非结构化金融数据挖掘

非结构化数据难题

文本多到没人读得完

每家美国上市公司每年申报一份 10-K、每季度申报一份 10-Q，多数大盘股还会在发布当天或一两天内召开财报电话会。 放到罗素 3000 指数的成分股上，这意味着每个财报季都有数千份文件和会议记录在相互重叠的时间窗口里落地——每一份都是几十页到几百页。无论规模多大的研究部门，都不可能在市场对数字做出反应之前把这些内容全部细读一遍。

Two Sigma Research 和 Point72 旗下的 Cubist Systematic Strategies 都公开谈过它们如何把这看作一种结构性的低效： 文本是公开的，但真正制造出机会的，是认真读完它的边际成本——对每一只标的、每一个季度、都在发布后几分钟内完成。 LLM 不会疲劳，不会跳过无聊的段落，给当天第四十份文件打分时和给第一份时一样仔细。

变化在哪里：LLM 与传统 NLP 的差别

词袋模型和基于词典的情绪打分（比如 Loughran-McDonald 词表）从 2000 年代起就用在学术金融研究里，但它们很粗糙——分不出"不令人失望"和"令人失望"。 在金融文本上微调过的 Transformer 模型（FinBERT 及其后继者）能理解否定、模糊措辞和上下文，这是数词频的方法永远做不到的。机构交易台如今大规模跑的，正是这一项具体的能力升级。

准备发言与问答环节：信号藏在哪里

照稿念的部分能告诉你的比你以为的少

财报电话会开头的"准备发言"部分是写好的，经过法务和投资者关系部门审核，在 CEO 和 CFO 开口之前就演练过了。 无论这个季度实际表现如何，它都被打磨得听起来自信、口径统一。关于财报电话会语言学的学术研究（延续 2010 年代初以来会计与金融文献中流行起来的工作）反复发现， 没有讲稿的分析师问答环节所携带的增量信息要多得多 相比准备好的部分——因为高管无法为自己控制不了的问题事先写好答案。

看变化量，而不只是看绝对水平

单场电话会的语调绝对分数，远不如它相对于同一管理团队自身历史基准的差值、以及相对于同一场会议中准备发言语调的差值来得重要。 自信的准备发言与充满模糊措辞的问答回答之间出现巨大的负向落差——有时被称为"语调背离"——比单独读任何一个环节都更具体：它说明管理层对写好的叙事很自在，但在没有讲稿的追问下明显没那么自在地为它辩护。

10-Q/10-K 语调提取：MD&A 与风险因素

信号在申报文件中集中在哪里

10-K 或 10-Q 大部分是套话——会计披露、标准法律措辞、SEC 规则要求的条目编号。 机构 NLP 研究始终盯住的是两个章节： 管理层讨论与分析（MD&A），这里管理层用自己的话解释业绩；以及 风险因素 章节，这里逐年新增、删除和改写的内容会被当成独立于情绪分数之外的另一种信号来跟踪。

同比相似度本身就是一个信号

用文本相似度指标把本季度的 MD&A 和风险因素文本与去年同期的申报文件做比对（对句向量算余弦相似度是散户可及的标准做法），能暴露出单看情绪分数会漏掉的实质性变化。 一份文本上与去年相似度超过 95%（含套话在内）的申报文件，说明管理层觉得没必要多说什么新东西——这本身就有信息含量，尤其是在经营层面确实发生了变化的季度。

情绪衰减曲线：半衰期难题

为什么"判断正确"的情绪还不够

一个语调信号方向可以是对的，但如果等你能动手时它已经过期，那它对交易毫无价值。 关于财报电话会和申报文件情绪的公开学术与业界研究一致发现，市场对文本语调的价格反应是前置的——大部分增量信息在发布后数小时（而不是数天）内就被吸收进价格，之后留下一条又长又薄的残余漂移尾巴，在随后几个交易日里逐渐消退。

把衰减建模成指数函数

把衰减曲线落地的标准做法，是把信号的可用权重当成"距发布经过时间"的指数函数，这样交易系统就能决定对一个"变旧但还没死"的信号还该给多少信心，而不是把它当成非黑即白的二选一（可交易/不可交易）：

weight(t) = 2^(-t / half_life)

其中：
  t          = 距申报文件/电话会发布已经过的小时数
  half_life  = 信号损失一半价值所需的估计小时数
               （示意：3-24 小时，取决于标的/流动性）

卫星与网络抓取的印证信号

当文本语调与现实世界数据吻合时，它更有力

Two Sigma Research 和 Point72/Cubist 都谈到过使用另类数据——卫星影像、抓取来的网页价格和招聘信息、应用使用遥测数据——它们不是独立信号，而是用来印证或反驳语调模型从文本中提取出的结论。 一段自信的 MD&A 加上与之吻合的卫星停车场车流数据或招聘岗位增长，合起来比任何一方单独出现都更强；而一段自信的 MD&A 配上相互矛盾的现实世界数据，本身也是信息——它标记出一份语调可能与现实不符的申报文件。

Python 实现：完整的 NLP 打分流水线

完整实现： 会议记录/申报文件摄取、章节切分（准备发言与问答、MD&A 与风险因素）、FinBERT 情绪打分、同比相似度、按衰减加权的信号汇总。

完整 Python 代码

"""
LLM alpha：非结构化金融数据挖掘流水线
面向散户改造的 FinBERT 语调提取，用于申报文件与财报电话会

作者：Plan My Retire
日期：2026 年 3 月
"""

import re
import numpy as np
import pandas as pd
from datetime import datetime
from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
from sklearn.metrics.pairwise import cosine_similarity
from sentence_transformers import SentenceTransformer


class FinancialTextScorer:
    """
    使用 FinBERT 系列模型为金融文本（会议记录、申报文件）打情绪分，
    并按章节做加权。
    """

    def __init__(self, model_name="ProsusAI/finbert"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.model.eval()
        self.embedder = SentenceTransformer("all-MiniLM-L6-v2")
        self.labels = ["positive", "negative", "neutral"]

    def score_passage(self, text, max_length=512):
        """
        为一段文本返回 [-1, 1] 区间内的带符号情绪分数。
        过长的段落会被切块后取分数平均。
        """
        chunks = self._chunk_text(text, max_length)
        if not chunks:
            return 0.0

        scores = []
        for chunk in chunks:
            inputs = self.tokenizer(chunk, return_tensors="pt", truncation=True,
                                     max_length=max_length)
            with torch.no_grad():
                logits = self.model(**inputs).logits
            probs = torch.softmax(logits, dim=1).numpy()[0]
            # 正面减负面，忽略中性部分的概率质量
            signed = probs[self.labels.index("positive")] - probs[self.labels.index("negative")]
            scores.append(signed)

        return float(np.mean(scores))

    def _chunk_text(self, text, max_length):
        words = text.split()
        chunk_size = max_length - 20  # 为特殊 token 预留空间
        return [" ".join(words[i:i + chunk_size])
                for i in range(0, len(words), chunk_size)] if words else []

    def embedding_similarity(self, text_a, text_b):
        """两段文本之间的余弦相似度，用于申报文件的同比对比。"""
        emb = self.embedder.encode([text_a, text_b])
        return float(cosine_similarity([emb[0]], [emb[1]])[0][0])


class EarningsCallSplitter:
    """
    利用标准的会议记录标记，把原始的财报电话会记录切分成
    准备发言和分析师问答两个部分。
    """

    QA_MARKERS = [
        r"question-and-answer session",
        r"questions? and answers?",
        r"we will now (begin|open) the (question|q&a)",
        r"first question",
    ]

    def split(self, transcript_text):
        lower = transcript_text.lower()
        split_idx = None
        for pattern in self.QA_MARKERS:
            match = re.search(pattern, lower)
            if match:
                split_idx = match.start()
                break

        if split_idx is None:
            # 兜底：如果找不到标记，就假定最后三分之一是问答环节
            split_idx = int(len(transcript_text) * 0.66)

        return {
            "prepared_remarks": transcript_text[:split_idx],
            "qa_section": transcript_text[split_idx:]
        }


class FilingSectionExtractor:
    """
    利用 SEC 标准的条目编号标题，从 10-K/10-Q 文本中
    抽取 MD&A 和风险因素章节。
    """

    SECTION_PATTERNS = {
        "risk_factors": r"item\s+1a\.?\s+risk factors(.*?)item\s+1b",
        "mda": r"item\s+7\.?\s+management.s discussion(.*?)item\s+7a",
    }

    def extract(self, filing_text):
        lower = filing_text.lower()
        sections = {}
        for name, pattern in self.SECTION_PATTERNS.items():
            match = re.search(pattern, lower, re.DOTALL)
            sections[name] = filing_text[match.start():match.end()] if match else ""
        return sections


class DecayWeightedSignal:
    """
    把原始语调分数与一个基于"距发布经过小时数"的指数衰减权重结合起来，
    避免把变旧的信号当成和新鲜信号一样可交易。
    """

    def __init__(self, half_life_hours=8.0):
        self.half_life = half_life_hours

    def weight(self, hours_elapsed):
        if hours_elapsed < 0:
            return 0.0
        return 2 ** (-hours_elapsed / self.half_life)

    def weighted_score(self, raw_score, release_time, as_of_time=None):
        as_of_time = as_of_time or datetime.now()
        hours_elapsed = (as_of_time - release_time).total_seconds() / 3600
        return raw_score * self.weight(hours_elapsed)


class TonePipeline:
    """
    端到端流水线：输入原始的会议记录/申报文件文本，
    输出按衰减加权的综合语调分数。
    """

    def __init__(self, half_life_hours=8.0):
        self.scorer = FinancialTextScorer()
        self.call_splitter = EarningsCallSplitter()
        self.filing_extractor = FilingSectionExtractor()
        self.decay = DecayWeightedSignal(half_life_hours)

    def score_earnings_call(self, transcript_text, release_time, as_of_time=None):
        sections = self.call_splitter.split(transcript_text)
        prepared_score = self.scorer.score_passage(sections["prepared_remarks"])
        qa_score = self.scorer.score_passage(sections["qa_section"])

        # 问答环节权重更高——增量信息含量更大
        composite_raw = 0.35 * prepared_score + 0.65 * qa_score
        divergence = prepared_score - qa_score

        return {
            "prepared_remarks_score": prepared_score,
            "qa_score": qa_score,
            "tone_divergence": divergence,
            "composite_raw": composite_raw,
            "composite_decay_weighted": self.decay.weighted_score(
                composite_raw, release_time, as_of_time
            ),
        }

    def score_filing(self, filing_text, prior_year_filing_text, release_time,
                      as_of_time=None):
        sections = self.filing_extractor.extract(filing_text)
        prior_sections = self.filing_extractor.extract(prior_year_filing_text)

        mda_score = self.scorer.score_passage(sections["mda"])
        risk_similarity = self.scorer.embedding_similarity(
            sections["risk_factors"], prior_sections["risk_factors"]
        )

        return {
            "mda_score": mda_score,
            "risk_factors_yoy_similarity": risk_similarity,
            "risk_factors_changed": risk_similarity < 0.85,  # 经验阈值
            "composite_decay_weighted": self.decay.weighted_score(
                mda_score, release_time, as_of_time
            ),
        }


# 用法示例
if __name__ == "__main__":
    pipeline = TonePipeline(half_life_hours=8.0)

    sample_transcript = """
    Thank you for joining us today. We are pleased to report strong
    results this quarter, with revenue growth accelerating...
    [prepared remarks continue]
    Question-and-Answer Session
    Analyst: Can you walk us through the margin pressure in the segment?
    CFO: Well, I think, um, there are some puts and takes there, and
    we're cautiously monitoring the situation going into next quarter...
    """

    release_time = datetime(2026, 3, 12, 16, 30)  # 下午 4:30 的电话会
    as_of = datetime(2026, 3, 12, 20, 0)  # 3.5 小时后打分

    result = pipeline.score_earnings_call(sample_transcript, release_time, as_of)

    print(f"Prepared remarks score: {result['prepared_remarks_score']:.3f}")
    print(f"Q&A score: {result['qa_score']:.3f}")
    print(f"Tone divergence (prepared - Q&A): {result['tone_divergence']:.3f}")
    print(f"Decay-weighted composite: {result['composite_decay_weighted']:.3f}")

散户落地：你实际落在曲线的哪一段

别再去抢头三个小时

拥有直连会议记录数据流和专用推理基础设施的低延迟机构交易台，占据了发布后的头几个小时——对一条跑在延迟数据和消费级算力上的散户流水线来说，衰减曲线的这一段不是一个现实的目标。 这个策略中散户可行的版本，瞄准的是 持续数天的残余漂移尾部，在这一段里，行动较慢的资金还在调仓，信号虽然更小，但延迟状态下更容易够得着。

聚焦覆盖不足的标的

由于分析师覆盖稀薄的标的衰减半衰期更长，散户做语调提取在中小盘股上的风险回报，在结构上要好于覆盖最密集的超大盘股，在后者身上，第一个小时的机构反应通常已经在散户流水线动手之前吃掉了大部分可用的价格影响。

示意性业绩

下面的数字是对一个瞄准 1-3 天残余窗口、按衰减加权的语调背离策略收益形态的假设性示意 ——它们不是业绩记录，也不承诺能取得类似结果。列出它们是为了把风险回报的形态说具体，而不是暗示这件事可以预测。

抹掉这个优势的常见错误

你的行动计划

第 1 阶段：搭建并验证打分流水线（4-8 周）

时间安排： 在投入任何资金之前，先确认这条流水线产出的分数合理且稳定。

1. 装好 FinBERT（Hugging Face，免费），拿它跑一批历史会议记录 这些记录之后的价格反应是已知的
2. 搭好准备发言/问答的切分器 并人工抽查一部分切分结果的准确性
3. 为每只标的计算并记录一条语调基准线 跨若干个季度积累，之后再把任何单次读数当成信号

第 2 阶段：小规模、考虑衰减的落地（2-4 个月）

时间安排： 从可交易的最小仓位起步，并且只做 1-3 天这个窗口。

1. 只交易残余尾部这个窗口 ——彻底跳过当日反应；你没有装备去争夺它
2. 优先选择覆盖不足的中小盘股 它们的衰减半衰期更长，散户的延迟影响更小
3. 要求语调背离与另类数据相互吻合 之后再决定仓位大小——绝不要仅凭文本语调就下满仓
4. 记录每一笔交易实际的衰减表现 以便按流动性分档细化你的半衰期估计

第 3 阶段：系统性扩大规模（6 个月以上）

时间安排： 必须先完整走过一个（或更多）财报周期的小仓位验证执行之后再说。

1. 把申报文件/会议记录的摄取自动化 对接实时的 EDGAR 提醒数据流和付费的会议记录供应商
2. 分散到一篮子标的上 而不是集中押注少数几只高信心个股
3. 每个财报季都重新验证一次衰减半衰期的估计 ——市场环境和覆盖格局会随时间变化

推荐阅读

• 机构研究：
  • Two Sigma Research——关于系统化投资中 NLP 与非结构化数据的公开评论
  • Point72 / Cubist Systematic Strategies——关于机器学习与另类数据流水线的公开评论
• PMR 相关文章：
  • Goldman Sachs 另类数据（配套文章，讲 FinBERT 与卫星/社交信号的集成）
  • Point72 Cubist 机器学习流水线（配套文章，讲这个语调信号将要接入的更完整的机器学习生产流水线）