散户情绪不是玄学：Python 爬东方财富股吧并量化的方法

原创

小白学大数据

发布于 2026-07-29 16:59:50

660

很多人一听到「散户情绪」就觉得是玄学，仿佛只有盘面高手靠直觉才能感知。其实东方财富股吧里每天几百万条发帖，就是最原始的散户情绪池子。帖子里有人喊「满仓干」，有人骂「这票废了」，有人发长长的复盘贴。把这些内容按条数抓下来，再用一套简单的规则给每条打分，散户情绪就能从感觉变成数字。举个真实的例子。某只票当天跌停，股吧前排热帖大概率是一堆「割肉了」「被套牢」的抱怨，这时候平均情绪分掉到负 0.6 以下很正常；过两天出个利好公告，热帖画风一转，满屏「加仓」「看好反转」，分数又能弹回正区间。你不需要读懂每一句，只要把这种整体摆动量出来，趋势就出来了。这篇文章就做两件事。第一，用 Python 按你指定的条数精确抓取东方财富股吧的帖子，抓满就停，不浪费请求。第二，用词典法给每条帖子算一个情绪分，最后聚合成一个能看的情绪指标。代码里会带上类型注解、dataclass、日志记录和重试策略，都是真正能跑的工程写法。先找到股吧的接口打开任意一只股票的股吧页面，比如贵州茅台（600519），按 F12 打开开发者工具，切到 Network 面板，刷新页面，筛选 XHR 请求。你会看到一个名为 Get 的接口，大致长这样：

https://guba.eastmoney.com/api/Get?code=600519&page=1&page_size=20&sort=0

code 是股票代码，page 是页码，page_size 是每页条数，sort=0 表示按时间排序。返回的是一段 JSON，帖子数组在 list 字段里，每条包含 post_id、post_content、post_publish_time、user_nickname 等字段。拿到这个接口，爬虫的骨头就出来了。有一点要提前说：返回的 post_content 经常夹着 HTML 标签和图片占位符，比如 <img src="...">、  之类。直接拿去打分会把噪声算进去。抓到内容后先做一次清洗，把标签剥掉，只留文字：

import re


def clean_html(raw: str) -> str:
    text = re.sub(r"<[^>]+>", "", raw)
    text = re.sub(r"&[a-z]+;", "", text)
    return text.strip()

清洗放在 fetch_posts 里塞进 GubaPost.content 之前，这样后面所有环节拿到的都是纯文本，情绪词典统计才准。还有个坑，post_id 在不同接口里有时是字符串有时是整数，统一转成 str 能少踩不少 Equal 比较的坑。先把数据结构定好用 dataclass 把单条帖子框住，后面无论是存盘还是打分，都围绕这个结构转，不至于在字典里乱翻 key。

from dataclasses import dataclass, asdict
from typing import Optional


@dataclass
class GubaPost:
    post_id: str
    stock_code: str
    nickname: str
    content: str
    publish_time: str
    sentiment: float = 0.0

    def to_dict(self) -> dict:
        return asdict(self)

sentiment 字段先给个默认值 0.0，等会儿打完分再回填。这样帖子模型和情绪计算解耦，逻辑清楚。配置日志和代理爬取过程中如果不打日志，出了问题你连哪页挂了都不知道。直接上标准库 logging，别用一堆 print。

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger("guba_crawler")

代理这块要单独说。股吧对高频访问不是完全没反应，单 IP 猛抓几百页很容易被限流。做批量抓取时，稳定的代理能省掉很多麻烦。我习惯用亿牛云的 HTTP 代理，它走用户名密码鉴权，代码里这样挂：

proxy_meta = {
    "host": "http://proxy.16yun.cn",
    "port": "31000",
    "user": "你的亿牛云用户名",
    "password": "你的亿牛云密码",
}

proxy_url = f"http://{proxy_meta['user']}:{proxy_meta['password']}@{proxy_meta['host'].split('//')[1]}:{proxy_meta['port']}"
proxies = {"http": proxy_url, "https": proxy_url}

把 proxies 透传给 requests 就行。亿牛云按量计费，平时小规模抓几十页用不上，真要跑几千条时它能把 IP 干净这件事兜住。抓之前还有个格式细节。东方财富的股票代码带市场前缀规则：沪市主板是 60xxxx，深市主板是 00xxxx，创业板是 30xxxx，科创板是 688xxx，北交所是 8xxxxx。股吧接口认的就是这六位纯数字 code，前面不加 SH、SZ 那种前缀。要抓多只票，把这些代码丢进一个列表循环调用 fetch_posts 即可，每只票单独存一份 CSV，后面拼表时按 stock_code 区分。按条数精准抓取核心需求是按条数抓，而不是无脑翻到底。下面这个函数接收目标条数，每翻一页把帖子累加，够了就 break，避免多抓浪费配额。每翻一页之间加一个随机间隔，比固定 sleep 更像真人，也能避免把请求节奏锤得太规律被识别。用 random.uniform 在 1 到 3 秒之间抖一下：

import random


def fetch_posts(stock_code: str, target_count: int, proxies: dict) -> List[GubaPost]:
    posts: List[GubaPost] = []
    page = 1
    page_size = 20

    while len(posts) < target_count:
        params = {
            "code": stock_code,
            "page": page,
            "page_size": page_size,
            "sort": 0,
        }
        try:
            resp = requests.get(BASE_URL, params=params, proxies=proxies, timeout=10)
            resp.raise_for_status()
            items = resp.json().get("list", [])
        except requests.RequestException as exc:
            logger.warning("第 %s 页抓取失败: %s", page, exc)
            break

        if not items:
            logger.info("已经没有更多帖子，第 %s 页为空", page)
            break

        for item in items:
            if len(posts) >= target_count:
                break
            raw = item.get("post_content", "")
            posts.append(
                GubaPost(
                    post_id=str(item.get("post_id", "")),
                    stock_code=stock_code,
                    nickname=item.get("user_nickname", ""),
                    content=clean_html(raw),
                    publish_time=item.get("post_publish_time", ""),
                )
            )

        logger.info("已抓取 %s/%s 条", len(posts), target_count)
        page += 1
        time.sleep(random.uniform(1, 3))

    return posts[:target_count]

这里有个细节：内层循环也判断了 len(posts) >= target_count，因为一页 20 条，目标可能是 35 条，第 2 页抓到 40 条时多出来的 5 条得切掉，否则就超了。最后 return posts[:target_count] 再兜一层，确保数量精确。另外注意 content 在入模前就过了 clean_html，前面说的标签噪声在这一步清掉，后面打分拿到的全是纯文字。给请求加重试网络抖动、代理偶发超时都正常，直接抛异常中断整个任务不划算。写个简单的重试装饰器，失败就退避几秒再试。

import time
from functools import wraps
from typing import Callable, TypeVar

T = TypeVar("T")


def retry(times: int = 3, delay: float = 2.0) -> Callable:
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> T:
            last_exc: Exception = Exception("unknown error")
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except requests.RequestException as exc:
                    last_exc = exc
                    logger.warning("第 %s 次重试，%s 秒后继续", attempt, delay)
                    time.sleep(delay * attempt)
            raise last_exc
        return wrapper
    return decorator

把 retry() 套到真正的请求函数上，前一次示例里的 requests.get 单独抽出来即可。退避时间按 delay * attempt 递增，避免瞬间疯狂重试把对面打毛。把情绪量化出来量化不一定要上深度学习。对股吧这种口语化、情绪浓的文本，词典法又快又解释得清。先准备两份词表：

POS_WORDS = ["涨", "利好", "加仓", "看好", "抄底", "拉升", "反弹", "盈利", "赚", "满仓"]
NEG_WORDS = ["跌", "利空", "割肉", "看空", "套牢", "暴跌", "亏损", "雷", "崩", "清仓"]


def score_sentiment(text: str) -> float:
    pos = sum(text.count(w) for w in POS_WORDS)
    neg = sum(text.count(w) for w in NEG_WORDS)
    total = pos + neg
    if total == 0:
        return 0.0
    return round((pos - neg) / total, 3)

score_sentiment 返回 -1 到 1 之间的数，1 代表全是正面词，负 1 代表全是负面词，0 表示没命中任何词。用 str.count 而不是分词，是为了轻量，股吧帖子短、口语多，逐字统计反而更稳。词典法有个绕不开的短板：它认字不认语境。「不跌」「没利空」这种带否定的表达，会被 str.count 当成正面算进去，分数就飘了。帖子短、否定句占比不高，整体指标不会被带偏太多，但想更准可以加一层简单处理：把内容按句号断句，句子里同时出现否定词（不、没、别）和情绪词时，把这条情绪词的权重翻负。改起来就几行：

NEG_MARKERS = ["不", "没", "别", "无"]


def score_sentiment_v2(text: str) -> float:
    pos = neg = 0
    for sent in re.split(r"[。！？\n]", text):
        has_neg = any(m in sent for m in NEG_MARKERS)
        for w in POS_WORDS:
            if w in sent:
                pos += -1 if has_neg else 1
        for w in NEG_WORDS:
            if w in sent:
                neg += -1 if has_neg else 1
    total = pos + neg
    if total == 0:
        return 0.0
    return round(pos / total, 3)

这么改之后，「这不跌才怪」这种反讽还是会算错，但那种属于人读都费劲的句子，先放过它。工程上做到 80 分够用，剩下的 20 分交给更大的语料和模型，不在这篇文章的范围内。跑完抓取后回灌分数：

def enrich_sentiment(posts: List[GubaPost]) -> List[GubaPost]:
    for post in posts:
        post.sentiment = score_sentiment(post.content)
    return posts

聚合成一个能看的指标单条分数没意义，聚起来才有用。比如算平均情绪、正负帖占比，再存成 CSV 方便后续画图或接进你自己的看板。

utf-8-sig 编码是为了让 Excel 直接打开中文不乱码，这个坑踩过一次就记住了。avg_sentiment 持续为正且走高，说明散户整体偏乐观；一直为负，说明恐慌情绪在堆积。把这个值按天画折线，你就能直观看到一只股票的情绪拐点，比盯着单条帖子靠谱。把情绪分画出来CSV 躺在硬盘里还是数字，画一张图才看得清走势。用 matplotlib 读 CSV，按发布时间排序，把每条帖子的 sentiment 点出来，再叠一条当日均值线：

import csv
from datetime import datetime
import matplotlib.pyplot as plt
from matplotlib.dates import date2num
def plot_sentiment(csv_path: str) -> None:
 times, scores = [], []
 with open(csv_path, encoding="utf-8-sig") as f:
 for row in csv.DictReader(f):
 try:
 t = datetime.strptime(row["publish_time"][:19], "%Y-%m-%d %H:%M:%S")
 except ValueError:
 continue
 times.append(date2num(t))
 scores.append(float(row["sentiment"]))
 plt.scatter(times, scores, s=12, alpha=0.5, label="单帖情绪")
 plt.axhline(sum(scores) / len(scores), color="red", label="均值")
 plt.legend()
 plt.show()

散点能看出当天情绪的离散程度：点全挤在正值区，说明一边倒看多；点上下乱蹦，说明多空分歧大，这种日子往往对应放量震荡。均值红线往哪边偏，就是当天散户的整体立场。把多天的数据按 stock_code 拼起来，就能画出一只票的情绪曲线，和 K 线叠在一起看，常常比技术指标更早反映情绪拐点。把抓取定时跑起来单日快照只能看个截面，情绪真正有价值的地方在变化。让脚本每天固定时间跑一次，把当天的平均情绪分追加进同一张表，时间一长就自然形成序列。不用上复杂调度框架，一个无限循环加 time.sleep 就能先跑起来：

import time
from datetime import datetime
def daily_job(stock_code: str, target: int, out_csv: str) -> None:
 while True:
 posts = fetch_posts(stock_code, target, proxies)
 posts = enrich_sentiment(posts)
 summary = summarize(posts, out_csv)
 logger.info("%s 任务完成: %s", datetime.now().date(), summary)
 time.sleep(86400)

正式环境建议交给系统的 cron 或任务计划程序，比进程里死循环省心，进程挂了也不会丢调度。跑之前记得把 time.sleep 的间隔、目标条数按你的代理配额定好，别一觉醒来把当月流量吃完。为什么非要按条数抓最后回头说一句按条数这件事。很多人写爬虫习惯「抓到没有为止」，对股吧这种信息流站点，全量抓既没意义又烧代理。你做情绪面板，每天同口径抓 200 条，前后才可比；今天抓 200 明天抓 5000，两条均值摆在一起会自己骗自己。把 target_count 钉死，等于给数据加了把尺子，后面所有分析都站得住脚。几点实在的提醒抓取频率别太猛，每页之间 time.sleep 一个一两秒的随机间隔，配合代理基本不会被拦。词表建议按板块自己补，比如新能源和白酒的「黑话」差很多，通用词典打出来会偏。另外股吧数据只反映发言的那部分人，沉默的大多数不在里面，用它做参考可以，拿来当买卖依据就过头了。这套流程跑通后，你完全可以把多只股票并起来抓，做成自己的散户情绪面板。代码量不大，真正的工作量在后面维护词表和看数据，但第一步把情绪变成数字，事情就从玄学变成了可以聊的东西。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python