Python抓市场情绪:AI做空时代的散户护身符

2025年7月3日，印度证券交易委员会（SEBI）对华尔街量化巨头 Jane Street 开出重磅罚单：冻结 $5.66亿 资产，指控其利用高频算法操纵印度衍生品市场。同期的官方数据显示，印度近1000万散户在衍生品市场全年亏损高达 1.05万亿卢比（约$126亿），平均每位散户亏损 11万卢比。

这不是孤立事件。全球范围内，量化基金正在用AI和算法从散户手中高效"收割"。当机构用每秒数千次的程序化交易做空时，散户能做什么？答案至少包括：用代码监控市场情绪，不让自己成为最后一个接盘的人。 本文手把手教你用Python搭建一套简易的市场情绪监控系统。

第一步：安装所需库

pip install requests beautifulsoup4 nltk matplotlib

四个库的职责分别是：requests 负责发送HTTP请求并获取网页源码，beautifulsoup4 负责解析HTML并提取结构化内容，nltk 提供VADER情感分析引擎（专门优化了社交媒体和短文本场景），matplotlib 则用于绘制可读的情绪趋势图表。如果你的环境中还没有这些库，用上面的命令一键安装即可。

如果安装 nltk 后首次运行时下载 VADER 词典较慢，可以预先手动下载：

第二步：编写新闻标题抓取器

我们从财经网站抓取最新的市场新闻标题作为情绪分析素材。需要特别说明的是：直接抓取商业网站的数据仅适用于个人学习和研究——生产级项目建议通过官方API（如NewsAPI、Alpha Vantage）或RSS源获取数据，避免因网页结构变化导致抓取失败，也减少合规风险。

以下是标题抓取器的完整实现：

import requests
from bs4 import BeautifulSoup

def fetch_headlines(url: str) -> list[str]:
    """抓取指定URL页面中的所有h2/h3标题"""
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
    }
    try:
        resp = requests.get(url, headers=headers, timeout=10)
        resp.raise_for_status()
        soup = BeautifulSoup(resp.text, "html.parser")
        # 提取所有h2和h3标签的文本
        titles = soup.find_all(["h2", "h3"])
        return [h.get_text(strip=True) for h in titles if h.get_text(strip=True)]
    except requests.RequestException as e:
        print(f"抓取失败: {e}")
        return []

# 示例：抓取Reuters市场新闻
url = "https://www.reuters.com/markets/"
headlines = fetch_headlines(url)
print(f"共抓取到 {len(headlines)} 条标题")
for i, h in enumerate(headlines[:10], 1):
    print(f"{i}. {h}")

第三步：情感分析引擎

使用 NLTK 的 VADER（Valence Aware Dictionary and sEntiment Reasoner）对每条标题打分。VADER 是专门为社交媒体和短文本场景调优的情感分析器，它不依赖复杂的机器学习模型，而是基于大量人工标注的词典进行规则匹配。它的优势在于能正确处理否定词（"不好"会得到负分）、程度副词（"太棒了"比"不错"得分更高）、以及表情符号和标点的影响。

核心原理是 VADER 为每条文本返回一个 compound 分数，范围从 -1（极度负面）到 +1（极度正面）。我们对所有标题取平均值，就能得到整体市场情绪的方向和强度：

from nltk.sentiment import SentimentIntensityAnalyzer
import nltk

# 首次运行需要下载词典
nltk.download("vader_lexicon", quiet=True)
sia = SentimentIntensityAnalyzer()

def analyze_sentiment(headlines: list[str]) -> dict:
    """分析标题列表的整体情绪"""
    if not headlines:
        return {"average_score": 0, "label": "无数据", "details": []}

    scores = []
    details = []
    for h in headlines:
        compound = sia.polarity_scores(h)["compound"]
        scores.append(compound)
        details.append((h, compound))

    avg = sum(scores) / len(scores)

    # 确定情绪标签
    if avg > 0.15:
        label = "积极乐观"
    elif avg < -0.15:
        label = "消极悲观"
    else:
        label = "中性"

    return {"average_score": avg, "label": label, "details": details}

result = analyze_sentiment(headlines)
print(f"市场情绪: {result['label']}")
print(f"综合得分: {result['average_score']:.3f}")
print("\n情绪最极端的5条:")
sorted_details = sorted(result["details"], key=lambda x: abs(x[1]), reverse=True)
for headline, score in sorted_details[:5]:
    print(f"  [{score:+.2f}] {headline[:60]}...")

第四步：可视化情绪趋势

单次分析只能告诉你"此刻"的市场情绪，但趋势才是真正有价值的信号。一段持续下滑的情绪曲线可能预示着市场恐慌正在形成；而突然的情绪逆转（从悲观跳变到乐观）往往对应着重大的政策或事件冲击。

我们将多次采集的情绪得分绘制成折线图，并标注警报线来辅助判断。实际使用时，这部分应该与定时任务结合（如每30分钟采集一次），形成持续的情绪时间序列：

import matplotlib.pyplot as plt
from datetime import datetime

# 模拟多次采样数据（实际使用时用定时任务持续采集）
history = {
    "times": ["07-25 09:30", "07-25 11:00", "07-25 14:30",
              "07-26 09:30", "07-26 11:00", "07-26 14:30"],
    "scores": [0.12, -0.08, -0.35, -0.42, -0.28, -0.55]
}

def plot_sentiment_trend(times: list, scores: list):
    plt.figure(figsize=(10, 5))
    plt.plot(range(len(scores)), scores,
             marker="o", color="#2196F3", linewidth=2,
             markersize=8)

    # 标注危险区域
    plt.axhline(y=-0.3, color="red", linestyle="--",
                alpha=0.6, label="悲观警报线 (-0.3)")
    plt.axhline(y=0.3, color="green", linestyle="--",
                alpha=0.6, label="乐观警报线 (+0.3)")
    plt.axhline(y=0, color="gray", linestyle="-", alpha=0.3)

    plt.xticks(range(len(times)), times, rotation=45)
    plt.ylabel("情绪综合得分", fontsize=12)
    plt.title("市场情绪趋势监控", fontsize=14, fontweight="bold")
    plt.legend(loc="upper left")
    plt.grid(alpha=0.3)
    plt.tight_layout()
    plt.savefig("sentiment_trend.png", dpi=150)
    print("趋势图已保存为 sentiment_trend.png")

plot_sentiment_trend(history["times"], history["scores"])

第五步：设置自动警报

当情绪数据异常偏离时自动触发通知。我们设置了两级判断：长期均分超过阈值意味着整体情绪已连续多天异常；近期趋势快速变化则可能是突发事件的即时反映。

下面的代码实现了两级警报机制。在生产环境中，你可以将 print 替换为实际的推送通道——企业微信、钉钉或邮件通知都可以简单接入：

import json

ALERT_THRESHOLD = 0.3  # 情绪得分绝对值超过此值触发警报

def check_and_alert(scores: list[float], threshold: float = ALERT_THRESHOLD):
    """检查是否需要发出警报"""
    if not scores:
        return False

    avg = sum(scores) / len(scores)
    recent_trend = scores[-3:] if len(scores) >= 3 else scores
    recent_avg = sum(recent_trend) / len(recent_trend)

    # 综合判断：整体情绪 + 近期趋势
    if abs(avg) > threshold:
        direction = "悲观" if avg < 0 else "过度乐观"
        print(f"🚨 警报！市场情绪异常{direction}！")
        print(f"   长期均分: {avg:.3f}")
        print(f"   近期均分: {recent_avg:.3f}")
        # TODO: 接入邮件、钉钉或企业微信通知
        return True
    elif abs(recent_avg) > threshold * 1.2:
        direction = "恶化" if recent_avg < avg else "过热"
        print(f"⚠️ 预警！市场情绪正在{direction}！")
        print(f"   近期均分: {recent_avg:.3f}")
        return True

    return False

check_and_alert(history["scores"])

进阶方向

本教程搭建的是最基础版本。你可以沿着以下方向扩展：

• • 多源聚合：同时抓取 Reuters、Bloomberg、雪球、东方财富等多个来源
• • 中文分词优化：用 jieba 分词 + 金融领域词典，提升中文标题的分析精度
• • 定时任务：用 crontab 或 APScheduler 实现每小时自动采集
• • 回测验证：将情绪数据与历史股价对比，验证情绪指标的有效性

专业机构用AI做空，散户至少可以用Python监控情绪信号。数据不会让你次次做对，但能让你不成为最盲目那个。

🔥 关注LC智趣厅