如何用Python分析股票市场情绪？代码+数据+避坑指南全解析

import requests
import pandas as pd
import time
from datetime import datetime

# 东方财富股吧帖子列表接口
# 参数说明: 
#   page=1 页码, pagesize=50 每页条数
#   field=258 表示按时间排序, 其余字段控制返回的列
def fetch_guba_posts(stock_code="600519", page=1):
    url = "https://gbapi.eastmoney.com/list"
    params = {
        "code": stock_code,
        "p": page,
        "ps": 50,
        "type": "2",  # 2=全部帖子
        "sid": "", 
        "field": "258",  # 按发布时间排序
        "dpt": "wz.astock"
    }
    headers = {
        "user-agent": "mozilla/5.0 (windows nt 10.0; win64; x64)",
        "referer": f"https://guba.eastmoney.com/list,{stock_code}.html"
    }
    resp = requests.get(url, params=params, headers=headers, timeout=10)
    data = resp.json()
    
    rows = []
    for item in data.get("result", {}).get("posts", []):
        rows.append({
            "post_id": item.get("postid"),
            "title": item.get("post_title"),
            "content": item.get("post_content"),
            "publish_time": datetime.fromtimestamp(item.get("post_publish_time", 0)),
            "read_count": item.get("post_read_count"),
            "comment_count": item.get("post_comments_count")
        })
    return rows

# 采集最近10页, 注意检查翻页停止条件
stock_code = "600519"
all_posts = []
for page in range(1, 11):
    posts = fetch_guba_posts(stock_code, page)
    if not posts:
        break
    all_posts.extend(posts)
    time.sleep(2)  # 控制请求频率, 别打太狠

df = pd.dataframe(all_posts)
df = df.drop_duplicates(subset="post_id")  # 去重, 翻页可能重复
df.to_csv(f"{stock_code}_posts.csv", index=false, encoding="utf-8-sig")
print(f"抓取完成, 共 {len(df)} 条帖子")

# 对已采集的df做清洗
df = df.drop_duplicates(subset=["title", "publish_time"])

# 过滤广告: 关键词命中即丢弃
ad_keywords = ["加微信", "qq群", "代客理财", "会员招募", "私聊", "点击链接"]
mask = df["title"].str.contains("|".join(ad_keywords), na=false)
df = df[~mask]

# 过滤超短文本: 少于10个汉字视为无效喊单
import re
def chinese_chars(text):
    if not isinstance(text, str):
        return 0
    return len(re.findall(r"[\u4e00-\u9fa5]", text))

df["char_len"] = df["title"].str.cat(df["content"], sep=" ").apply(chinese_chars)
df = df[df["char_len"] >= 10]

# 重置索引, 避免后续处理时索引断裂
df = df.reset_index(drop=true)

# 金融情感词典的最小实现
# 正负向词表: 每个词带一个强度权重, 范围0.5~2.0
positive_words = {
    "涨停": 2.0, "利好": 1.5, "突破": 1.2, "反包": 1.5,
    "主升": 1.8, "超预期": 1.2, "加仓": 0.8, "抄底": 1.0
}
negative_words = {
    "跌停": -2.0, "利空": -1.5, "爆仓": -2.0, "套牢": -1.5,
    "割肉": -1.2, "退市": -2.0, "st": -1.0, "减持": -1.2,
    "跳水": -1.8, "核按钮": -2.0
}

import re
import snownlp

def financial_sentiment(text):
    # 先跑snownlp基线, 再按词典做覆盖修正
    base_score = snownlp.snownlp(text).sentiments
    # 打分为0~1, 转成-1~1便于对比
    base_score = (base_score - 0.5) * 2
    
    words = re.findall(r"[\u4e00-\u9fa5]{2,4}", text)
    dict_score = 0.0
    hit_count = 0
    for word in words:
        if word in positive_words:
            dict_score += positive_words[word]
            hit_count += 1
        elif word in negative_words:
            dict_score += negative_words[word]
            hit_count += 1
    
    # 如果词典强度超过基线, 直接采用词典分数
    if abs(dict_score) > abs(base_score) and hit_count >= 2:
        # 归一化到[-1, 1]区间
        final_score = max(-1.0, min(1.0, dict_score / (hit_count * 1.5)))
    else:
        final_score = base_score
    return final_score

# 使用transformers库微调macbert的示例骨架
# 需要安装: pip install transformers torch datasets
import torch
from transformers import autotokenizer, automodelforsequenceclassification, trainer, trainingarguments
from datasets import dataset

# 标注数据格式: text + label, label 0=负面 1=正面
train_data = [
    {"text": "今天终于回本了，清仓走人", "label": 1},
    {"text": "跌成狗了，管理层在干嘛", "label": 0},
    # ... 更多标注样本, 建议至少2000条
]

# 加载macbert中文模型
# 该模型基于哈工大开源的中文macbert预训练权重
model_name = "hfl/chinese-macbert-base"
tokenizer = autotokenizer.from_pretrained(model_name)
model = automodelforsequenceclassification.from_pretrained(model_name, num_labels=2)

def tokenize_fn(examples):
    # max_length=128: 股吧帖子标题加正文绝大多数在100字以内
    return tokenizer(examples["text"], truncation=true, padding="max_length", max_length=128)

dataset = dataset.from_list(train_data).map(tokenize_fn, batched=true)

training_args = trainingarguments(
    output_dir="./finetuned_sentiment",
    learning_rate=2e-5,
    per_device_train_batch_size=16,
    num_train_epochs=3,
    weight_decay=0.01,
    save_strategy="epoch",
)

trainer = trainer(
    model=model,
    args=training_args,
    train_dataset=dataset,
)
trainer.train()

import pandas as pd
import numpy as np

def aggregate_daily_sentiment(df):
    # df: 包含 post_time, sentiment_score, read_count 三列
    df = df.copy()
    df["date"] = pd.to_datetime(df["post_time"]).dt.date
    # 阅读数取对数压缩极大值
    df["weight"] = np.log1p(df["read_count"] + 1)
    
    # 按日期聚合: 加权均值
    daily = df.groupby("date").apply(
        lambda x: np.average(x["sentiment_score"], weights=x["weight"])
    ).reset_index(name="weighted_sentiment")
    
    # 正面/负面占比
    df["polarity"] = np.where(df["sentiment_score"] > 0.05, 1, 
                     np.where(df["sentiment_score"] < -0.05, -1, 0))
    polarity_ratio = df.groupby("date")["polarity"].value_counts(normalize=true).unstack()
    
    daily = daily.join(polarity_ratio, on="date")
    # bullish_bearish_spread: 正面占比减负面占比
    daily["spread"] = daily.get(1, 0) - daily.get(-1, 0)
    return daily

# 调用示例
daily_agg = aggregate_daily_sentiment(df)
print(daily_agg.tail())

# 合并行情数据: daily_agg 是上一步的结果
# price_df: 列包含 date, open, close, high, low
merged = daily_agg.merge(price_df, on="date", how="inner")
merged["next_ret"] = merged["close"].pct_change().shift(-1)  # t+1收益率
merged["ret"] = merged["close"].pct_change()                  # t日收益率

# 计算当前情绪与未来收益的相关性
from scipy.stats import pearsonr
corr_t1, p_value_t1 = pearsonr(merged["spread"], merged["next_ret"])
corr_t0, p_value_t0 = pearsonr(merged["spread"], merged["ret"])
print(f"当日情绪 vs 次日收益 相关系数: {corr_t1:.4f}, p值: {p_value_t1:.4f}")
print(f"当日情绪 vs 当日收益 相关系数: {corr_t0:.4f}, p值: {p_value_t0:.4f}")

merged["spread_zscore"] = (merged["spread"] - merged["spread"].rolling(20).mean()) / merged["spread"].rolling(20).std()
merged["signal"] = 0
# 情绪过热且当日涨幅偏大: 看空信号
merged.loc[(merged["spread_zscore"] > 1.5) & (merged["ret"] > 0.03), "signal"] = -1
# 情绪极冷且跌幅有限: 看多信号
merged.loc[(merged["spread_zscore"] < -1.5) & (merged["ret"] > -0.02), "signal"] = 1

import pandas as pd
import numpy as np
from scipy.stats import pearsonr

def backtest_sentiment_strategy(post_csv, price_csv, zscore_threshold=1.5):
    # 读数据
    df = pd.read_csv(post_csv, parse_dates=["post_time"])
    price = pd.read_csv(price_csv, parse_dates=["date"])
    
    # 1. 清洗
    # 2. 情感打分: 这里用简化版, 实际可替换成你的模型推理
    df["sentiment"] = df["title"].apply(financial_sentiment)
    
    # 3. 日度聚合
    daily = aggregate_daily_sentiment(df)
    
    # 4. 合并行情
    merged = daily.merge(price, on="date", how="inner")
    merged["ret"] = merged["close"].pct_change()
    merged["next_ret"] = merged["close"].pct_change().shift(-1)
    merged["zscore"] = (merged["spread"] - merged["spread"].rolling(20).mean()) / merged["spread"].rolling(20).std()
    
    # 5. 生成信号
    merged["signal"] = 0
    merged.loc[(merged["zscore"] < -zscore_threshold) & (merged["ret"] > -0.02), "signal"] = 1
    merged.loc[(merged["zscore"] > zscore_threshold) & (merged["ret"] > 0.03), "signal"] = -1
    
    # 6. 统计信号胜率: 次日收益与信号同向为胜
    valid = merged.dropna(subset=["signal", "next_ret"])
    win_long = (valid["signal"] == 1) & (valid["next_ret"] > 0)
    win_short = (valid["signal"] == -1) & (valid["next_ret"] < 0)
    hit_rate = (win_long.sum() + win_short.sum()) / len(valid)
    
    print(f"信号总数: {len(valid)}")
    print(f"胜率: {hit_rate:.2%}")
    print(f"次日平均收益: {valid['next_ret'].mean():.4%}")
    return merged