数据源准备：获取股票行情数据、新闻标题与社交媒体数据
做市场情绪分析，第一步就是搞定数据源。说白了，没有数据，再牛的模型也是空中楼阁。
我个人习惯把数据源分成三类：行情数据、新闻标题、社交媒体数据。这三类数据各有各的脾气，获取方式也完全不同。今天我就把我在项目中踩过的坑、总结的经验，一次性讲清楚。
核心观点：数据质量决定分析上限。宁可花80%的时间在数据清洗上，也不要急着跑模型。
一、整体数据架构
先看一张图，帮你快速建立全局认知。这是我做项目时画的框架，后来一直沿用。
这张图的核心逻辑很简单：三条数据流分别采集，经过清洗对齐后，统一输入情绪分析引擎。嗯，这里要注意，数据对齐往往是最大的坑，后面我会细说。
二、股票行情数据获取
行情数据是情绪分析的「锚点」。你得知道股价在涨还是跌，才能判断情绪是正面还是负面。
2.1 使用 tushare
tushare 是我最早用的库，数据全，但需要 token。我记得第一次用的时候，光申请 token 就折腾了半天。
import tushare as ts
# 设置 token（需要去官网注册获取）
ts.set_token('your_token_here')
pro = ts.pro_api()
# 获取日线行情
df = pro.daily(
ts_code='000001.SZ', # 平安银行
start_date='20240101',
end_date='20241231'
)
print(df.head())
小技巧：tushare 的 pro 接口有频率限制。我一般会加个 time.sleep(0.5)，避免被限流。
2.2 使用 akshare
akshare 是后来发现的宝藏库。免费、无需 token，而且数据源覆盖广。说白了，对个人开发者非常友好。
import akshare as ak
# 获取 A 股日线数据
df = ak.stock_zh_a_hist(
symbol="000001",
period="daily",
start_date="20240101",
end_date="20241231",
adjust="qfq" # 前复权
)
print(df.head())
避坑指南：我曾经在 akshare 上踩过一个坑——它返回的日期列是 object 类型，不是 datetime。记得手动转换：df['日期'] = pd.to_datetime(df['日期'])
2.3 两个库的对比
| 特性 | tushare | akshare |
|---|---|---|
| 是否需要 token | 是 | 否 |
| 数据频率限制 | 有（200次/分钟） | 无明确限制 |
| 数据覆盖范围 | A股、港股、美股、期货 | A股为主，部分期货 |
| 数据质量 | 高（官方合作） | 中（爬虫采集） |
| 适合场景 | 生产环境、机构使用 | 个人研究、快速原型 |
我个人建议：做研究用 akshare，做产品用 tushare。你想想看，生产环境里数据源突然断了，那可不是闹着玩的。
三、新闻标题数据获取
新闻标题是情绪的「风向标」。一条「突发利空」的标题，可能瞬间改变市场情绪。
3.1 财经新闻 API
我常用的方案是爬取东方财富、新浪财经的新闻列表。这里给一个简单的示例：
import requests
from bs4 import BeautifulSoup
import pandas as pd
def get_eastmoney_news(page=1):
"""获取东方财富快讯"""
url = f"https://finance.eastmoney.com/a/czqyw_{page}.html"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
resp = requests.get(url, headers=headers)
soup = BeautifulSoup(resp.text, 'html.parser')
news_list = []
for item in soup.select('.news-item'):
title = item.select_one('.title').text.strip()
link = item.select_one('a')['href']
news_list.append({'title': title, 'link': link, 'source': '东方财富'})
return pd.DataFrame(news_list)
# 获取最近10页新闻
all_news = pd.DataFrame()
for i in range(1, 11):
df = get_eastmoney_news(i)
all_news = pd.concat([all_news, df], ignore_index=True)
print(f"已获取第{i}页，共{len(df)}条")
print(f"总计获取 {len(all_news)} 条新闻标题")
关键点：新闻标题的时效性非常重要。我一般只取最近24小时内的新闻，太旧的数据对情绪分析没有意义。
3.2 数据清洗要点
新闻标题数据通常很脏。我在项目中遇到过这些问题：
- 重复标题：同一新闻被多个平台转载
- 特殊字符：如「【】★◆」等符号
- 无关内容：广告、推广信息
def clean_news_title(title):
"""清洗新闻标题"""
import re
# 去除特殊字符
title = re.sub(r'[【】★◆◆◆]', '', title)
# 去除多余空格
title = re.sub(r'\s+', ' ', title).strip()
# 去除过短的标题（可能是广告）
if len(title) < 10:
return None
return title
# 应用清洗
all_news['clean_title'] = all_news['title'].apply(clean_news_title)
all_news = all_news.dropna(subset=['clean_title'])
四、社交媒体数据获取
社交媒体是情绪的「放大器」。一条微博、一个帖子，可能引发连锁反应。
4.1 微博数据获取
微博的 API 现在比较严格。我常用的方法是模拟浏览器请求：
import requests
import json
def get_weibo_hot_search():
"""获取微博热搜榜"""
url = "https://weibo.com/ajax/side/hotSearch"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Cookie': '你的微博cookie' # 需要登录后获取
}
resp = requests.get(url, headers=headers)
data = resp.json()
hot_list = []
for item in data['data']['realtime']:
hot_list.append({
'title': item['word'],
'hot_value': item['raw_hot'],
'category': item.get('category', '综合')
})
return pd.DataFrame(hot_list)
# 获取热搜
hot_df = get_weibo_hot_search()
print(hot_df.head(10))
注意：微博的反爬虫机制很强。我曾经因为请求频率太高，被限制了三天。建议每次请求间隔至少 5 秒，并且使用代理 IP。
4.2 雪球数据获取
雪球是股民聚集地，情绪浓度极高。我一般获取热门帖子和评论：
import akshare as ak
def get_xueqiu_hot():
"""获取雪球热门帖子"""
# akshare 已经封装了雪球接口
df = ak.stock_social_hot_rank_xq(symbol="全部")
return df
# 获取热门帖子
xueqiu_df = get_xueqiu_hot()
print(xueqiu_df[['title', 'hot_score', 'comments_count']].head())
4.3 社交媒体数据的特殊性
社交媒体数据和行情数据有个本质区别：非结构化。你想想看，一条「卧槽，跌惨了」和「市场出现系统性风险」，表达的情绪可能完全不同。
我一般会做以下预处理：
- 文本长度过滤：太短的帖子（< 5字）通常没有分析价值
- 情感词典匹配：用基础的情感词典做初步分类
- 时间戳对齐：把社交媒体数据按分钟对齐到行情数据
五、数据对齐与存储
三条数据源的时间粒度不同：行情数据是日线/分钟线，新闻是小时级，社交媒体是分钟级。怎么对齐？
我常用的策略是：以行情数据的时间戳为基准，向前聚合。
import pandas as pd
def align_data(stock_df, news_df, social_df):
"""将新闻和社交媒体数据对齐到行情数据"""
# 确保日期格式一致
stock_df['date'] = pd.to_datetime(stock_df['date'])
news_df['date'] = pd.to_datetime(news_df['date'])
social_df['date'] = pd.to_datetime(social_df['date'])
# 按日期聚合新闻数量
news_agg = news_df.groupby(news_df['date'].dt.date).size().reset_index()
news_agg.columns = ['date', 'news_count']
# 按日期聚合社交媒体帖子数量
social_agg = social_df.groupby(social_df['date'].dt.date).size().reset_index()
social_agg.columns = ['date', 'social_count']
# 左连接合并
result = stock_df.merge(news_agg, on='date', how='left')
result = result.merge(social_agg, on='date', how='left')
# 填充缺失值
result['news_count'] = result['news_count'].fillna(0)
result['social_count'] = result['social_count'].fillna(0)
return result
# 执行对齐
aligned_df = align_data(stock_data, news_data, social_data)
print(aligned_df.head())
经验之谈：数据对齐时，不要只聚合数量。我还会计算「情感得分均值」、「情感波动率」等衍生特征，这些对后续的情绪建模非常有帮助。
六、完整数据流水线
最后，我把整个数据准备流程封装成一个类。这样每次跑新数据时，直接调用就行。
class MarketDataPipeline:
"""市场数据流水线"""
def __init__(self):
self.stock_data = None
self.news_data = None
self.social_data = None
def fetch_all(self, stock_code, start_date, end_date):
"""获取所有数据"""
print("开始获取行情数据...")
self.stock_data = self._fetch_stock(stock_code, start_date, end_date)
print("开始获取新闻数据...")
self.news_data = self._fetch_news(start_date, end_date)
print("开始获取社交媒体数据...")
self.social_data = self._fetch_social(start_date, end_date)
print("数据对齐中...")
result = align_data(self.stock_data, self.news_data, self.social_data)
return result
def _fetch_stock(self, code, start, end):
# 使用 akshare 获取行情
import akshare as ak
return ak.stock_zh_a_hist(symbol=code, start_date=start, end_date=end)
def _fetch_news(self, start, end):
# 获取新闻标题
# 实际项目中会调用爬虫
pass
def _fetch_social(self, start, end):
# 获取社交媒体数据
pass
# 使用示例
pipeline = MarketDataPipeline()
data = pipeline.fetch_all('000001', '20240101', '20241231')
print(f"最终数据形状: {data.shape}")
嗯，到这里，数据源准备的核心内容就讲完了。你可能会问：这些代码能直接跑吗？说实话，不能。因为每个数据源都需要你注册账号、获取 token 或 cookie。但框架和思路是通用的，你照着这个结构去填充具体实现就行。
数据准备是情绪分析中最枯燥、但最重要的一环。把这一步做扎实了，后面的模型构建就是水到渠成的事。
公众号:蓝海资料掘金营，微信deep3321