NLP情感分析与量化择时策略

市场由人组成，人由情绪驱动。如果能量化市场情绪，你就能在拥挤的羊群中提前看到转折。

情绪驱动市场的证据

行为金融学的洞见

有效市场假说      行为金融学
"价格反映所有信息"  "价格反映信息 + 情绪"
     ↓                 ↓
  随机游走          可预测的偏差

• 过度反应 → 恐慌性抛售 → 超卖机会
• 反应不足 → 利好消息缓慢消化 → 趋势延续
• 羊群效应 → 泡沫和崩盘的自增强循环

NLP 让我们有机会量化这些情绪。

NLP情绪分析技术栈

技术路线对比

方法	复杂度	准确率	可解释性	适用场景
词典法	低	60-70%	高	快速原型
传统ML	中	70-80%	中	中等数据量
FinBERT	中	80-88%	中	金融文本
GPT/Claude	高	85-92%	低	复杂语义
微调LLM	高	88-95%	低	领域定制

数据源分类

结构化数据源：
├── 财经新闻（Bloomberg, Reuters）
├── 公司公告和财报
├── 分析师研报
└── 央行政策声明

非结构化数据源：
├── 社交媒体（Twitter/X, Reddit, 雪球）
├── 论坛讨论（WallStreetBets, 股吧）
├── 搜索引擎趋势
└── 视频/播客转文字

实战一：基于 FinBERT 的情绪因子

FinBERT 快速上手

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
import pandas as pd
import numpy as np

# 加载 FinBERT（在金融文本上微调的BERT）
model_name = "ProsusAI/finbert"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(model_name)

def sentiment_analysis(texts, batch_size=32):
    """批量情感分析"""
    sentiments = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        inputs = tokenizer(
            batch, padding=True, truncation=True,
            max_length=512, return_tensors="pt"
        )

        with torch.no_grad():
            outputs = model(**inputs)
            probs = torch.softmax(outputs.logits, dim=-1)

        # FinBERT 标签: 0=负面, 1=中性, 2=正面
        sentiments.append(probs.numpy())

    return np.concatenate(sentiments, axis=0)

# 示例：分析一组新闻标题
headlines = [
    "Apple reports record quarterly earnings, beating analyst expectations",
    "Market tumbles amid rising inflation fears and Fed uncertainty",
    "Tesla faces regulatory investigation over autopilot safety concerns"
]

probs = sentiment_analysis(headlines)
for h, p in zip(headlines, probs):
    sentiment = ['负面', '中性', '正面'][np.argmax(p)]
    print(f"[{sentiment}] {h}")
    print(f"  负面:{p[0]:.1%} 中性:{p[1]:.1%} 正面:{p[2]:.1%}\n")

构建日频情绪因子

def build_daily_sentiment_factor(news_df, date_col='date',
                                  headline_col='headline'):
    """从新闻数据构建每日情绪因子"""

    # 对每天的所有新闻计算情绪
    daily_sentiments = []

    for date, group in news_df.groupby(date_col):
        texts = group[headline_col].tolist()
        probs = sentiment_analysis(texts)

        daily_sentiments.append({
            'date': date,
            'news_count': len(texts),
            'sentiment_positive': probs[:, 2].mean(),  # 正面均值
            'sentiment_negative': probs[:, 0].mean(),  # 负面均值
            'sentiment_score': probs[:, 2].mean() - probs[:, 0].mean(),  # 净情绪
            'sentiment_std': probs[:, 2].std(),  # 情绪分歧度
            'max_positive': probs[:, 2].max(),
            'max_negative': probs[:, 0].max(),
        })

    sentiment_df = pd.DataFrame(daily_sentiments)
    sentiment_df.set_index('date', inplace=True)

    # 衍生因子
    sentiment_df['sentiment_ma5'] = sentiment_df['sentiment_score'].rolling(5).mean()
    sentiment_df['sentiment_change'] = sentiment_df['sentiment_score'].diff()
    sentiment_df['sentiment_zscore'] = (
        (sentiment_df['sentiment_score'] - sentiment_df['sentiment_score'].rolling(20).mean())
        / sentiment_df['sentiment_score'].rolling(20).std()
    )

    return sentiment_df

实战二：用 GPT/Claude 做深度语义分析

Few-Shot 文本分类

import openai

SYSTEM_PROMPT = """你是一个专业的金融情绪分析系统。
分析给定的财经文本，输出JSON格式：

{
  "sentiment": "positive/negative/neutral",
  "confidence": 0.0-1.0,
  "market_impact": "high/medium/low",
  "key_factors": ["因素1", "因素2"],
  "sentiment_score": -1.0到1.0之间的数值
}

评分标准：
- 1.0: 强烈利好（超预期盈利、重大政策利好）
- 0.5: 温和利好
- 0.0: 中性或信息混杂
- -0.5: 温和利空
- -1.0: 强烈利空（暴雷、监管处罚）
"""

def deep_sentiment_analysis(text, model="gpt-4"):
    """用GPT进行深度情绪分析"""
    response = openai.ChatCompletion.create(
        model=model,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": text}
        ],
        temperature=0.1,  # 低温度保证一致性
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

财报电话会议分析

财报电话会议的语气和措辞往往比财报数字本身更能预测未来：

def analyze_earnings_call(transcript):
    """分析财报电话会议的文本"""

    prompt = f"""分析以下财报电话会议记录，重点关注：

1. 管理层语气变化（与过往对比）
2. 回避问题的次数和话题
3. 不确定性的语言指标（"可能"、"取决于"、"具有挑战性"等词频）
4. 对未来的具体承诺 vs 模糊表述

电话会议记录：
{transcript}

输出JSON：
{{
  "tone_score": -1到1（-1=非常消极，1=非常积极）,
  "evasiveness_score": 0到1（越高=越回避问题）,
  "uncertainty_index": 不确定性词汇密度,
  "concrete_vs_vague_ratio": 具体承诺/模糊表述的比值,
  "red_flags": ["红旗信号列表"],
  "summary": "两句话总结"
}}
"""
    return call_llm(prompt)

实战三：Reddit/社交媒体情绪

WallStreetBets 情绪数据

import requests
from datetime import datetime, timedelta

def fetch_reddit_sentiment(subreddit, ticker, days=7):
    """获取Reddit上某股票的情绪"""

    # 使用Pushshift API获取帖子
    end_date = datetime.utcnow()
    start_date = end_date - timedelta(days=days)

    url = f"https://api.pushshift.io/reddit/search/submission/"
    params = {
        'subreddit': subreddit,
        'q': ticker,
        'after': int(start_date.timestamp()),
        'before': int(end_date.timestamp()),
        'size': 500,
        'sort': 'score'
    }

    response = requests.get(url, params=params)
    posts = response.json()['data']

    # 提取文本和分数
    texts = [p.get('title', '') + ' ' + p.get('selftext', '') for p in posts]
    scores = [p.get('score', 0) for p in posts]
    comments = [p.get('num_comments', 0) for p in posts]

    # NLP分析
    sentiments = sentiment_analysis(texts)

    return pd.DataFrame({
        'text': texts,
        'score': scores,
        'comments': comments,
        'sentiment_positive': sentiments[:, 2],
        'sentiment_negative': sentiments[:, 0],
    })

散户情绪指标

def build_retail_sentiment_index(ticker):
    """构建散户情绪综合指标"""

    data = fetch_reddit_sentiment('wallstreetbets', ticker)

    # 加权情绪（高分帖子权重大）
    weighted_sentiment = np.average(
        data['sentiment_positive'] - data['sentiment_negative'],
        weights=np.log1p(data['score']) * np.sqrt(data['comments'] + 1)
    )

    # 情绪热度（讨论量）
    buzz_score = np.log1p(len(data))

    # 情绪一致性（分歧度）
    sentiment_dispersion = data[['sentiment_positive', 'sentiment_negative']].std().mean()

    return {
        'sentiment': weighted_sentiment,
        'buzz': buzz_score,
        'dispersion': sentiment_dispersion,
        'composite': weighted_sentiment * buzz_score / (1 + sentiment_dispersion)
    }

情绪因子与交易信号

择时策略框架

def generate_sentiment_signals(sentiment_df):
    """从情绪因子生成交易信号"""

    signals = pd.DataFrame(index=sentiment_df.index)

    # 极端情绪反转信号
    # 当情绪过于乐观（>2σ），可能见顶
    # 当情绪过于悲观（<-2σ），可能见底
    signals['extreme_bearish'] = sentiment_df['sentiment_zscore'] < -2.0  # 买入机会
    signals['extreme_bullish'] = sentiment_df['sentiment_zscore'] > 2.0   # 卖出警告

    # 情绪趋势变化
    # 从悲观转为中性/乐观 = 买入信号
    signals['sentiment_turnaround'] = (
        (sentiment_df['sentiment_zscore'] > -0.5) &
        (sentiment_df['sentiment_zscore'].shift(1) < -1.0)
    )

    # 情绪与价格背离（最强信号之一）
    # 价格下跌 + 情绪改善 = 潜在底部
    signals['bullish_divergence'] = (
        (sentiment_df['sentiment_change'] > 0) &
        (sentiment_df['price_change'] < 0)
    )

    # 情绪一致性过滤
    # 情绪分歧过大时，信号可信度降低
    signals['high_confidence'] = sentiment_df['sentiment_std'] < 0.15

    # 最终信号
    signals['buy_signal'] = (
        (signals['extreme_bearish'] | signals['sentiment_turnaround']) &
        signals['high_confidence']
    )

    signals['sell_signal'] = (
        signals['extreme_bullish'] & signals['high_confidence']
    )

    return signals

实战建议

关键要点

1. 多源融合 — 单一来源的情绪信号噪音太大，多源交叉验证
2. 语境重要 — “利润下降”在预期下降20%但实际降10%时是利好
3. 时间衰减 — 新闻情绪的影响随时间快速衰减（通常1-5天）
4. 极端值最有信息量 — 正常情绪波动预测力弱，极端情绪预测力强
5. 与价格结合 — 情绪+量价分析 > 单独情绪 > 单独量价

常见陷阱

- 所有新闻都是"中性偏正面" → 需要校准
- 机器难以理解讽刺和反讽 → Reddit情绪需要特殊处理
- 重大事件日新闻爆发 → 需要按事件去重而非按篇数
- 回译偏差 → 中文文本用中文模型（如FinBERT-Chinese）

---

市场情绪是可以量化的。NLP 给了我们一把读取市场”心情”的钥匙。但记住：当所有人都在用同一把钥匙时，锁就会换掉。不断寻找新的情绪数据源，是在这个领域保持优势的关键。