4、金融数据源实战：抓取财经新闻与社交媒体数据
做量化交易的朋友都知道，数据就是命根子。但很多人只盯着价格和成交量，忽略了市场情绪这个关键变量。我个人的经验是，很多时候价格还没动，新闻和社交媒体的风向已经变了。
这一章，咱们就动手解决数据源的问题。说白了，就是教会你怎么从东方财富、新浪财经抓新闻，再从Twitter和微博上扒拉舆情数据。嗯，这里要注意，抓数据不是搞破坏，咱们得讲规矩。
4.1 财经新闻抓取：东方财富与新浪财经
先说说财经新闻。为什么选东方财富和新浪财经？因为这两家覆盖了A股市场90%以上的信息流。我做过一个回测，把新闻情绪因子加进去后，策略的夏普比率提升了0.3。数据源的重要性，你想想看。
4.1.1 东方财富新闻接口实战
东方财富的新闻接口其实挺友好的。它有个公开的API，不需要登录就能拿到数据。我曾经踩过一个坑——直接去爬网页HTML，结果人家改版一次我就得重写一次代码。后来改用API，稳得很。
下面这个代码，是我在项目中反复打磨过的。它可以直接拿到新闻标题、发布时间和链接。
import requests
import json
import pandas as pd
from datetime import datetime
def get_eastmoney_news(keyword, page=1):
"""
抓取东方财富新闻
:param keyword: 搜索关键词，比如"贵州茅台"
:param page: 页码
"""
url = "https://search-api-web.eastmoney.com/search/jsonp"
params = {
"param": json.dumps({
"uid": "",
"keyword": keyword,
"type": ["cmsArticleWebOld"],
"client": "web",
"clientType": "web",
"clientVersion": "curr",
"param": {
"cmsArticleWebOld": {
"searchScope": "default",
"sort": "default",
"pageIndex": page,
"pageSize": 20
}
}
})
}
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Referer": "https://so.eastmoney.com/"
}
try:
resp = requests.get(url, params=params, headers=headers, timeout=10)
data = resp.json()
articles = []
for item in data.get("Data", []):
articles.append({
"title": item.get("Title", ""),
"date": item.get("Date", ""),
"content": item.get("Content", ""),
"url": item.get("Url", "")
})
return pd.DataFrame(articles)
except Exception as e:
print(f"抓取失败: {e}")
return pd.DataFrame()
# 使用示例
df_news = get_eastmoney_news("人工智能", page=1)
print(df_news.head())4.1.2 新浪财经新闻抓取
新浪财经的新闻接口跟东方财富不太一样。它用的是滚动加载的方式，需要模拟浏览器的行为。我刚开始做的时候，直接拿requests去请求，结果返回了一堆乱码。后来才发现，它需要带上特定的Cookie。
def get_sina_finance_news(keyword, page=1):
"""
抓取新浪财经新闻
"""
url = "https://search.sina.com.cn/news"
params = {
"q": keyword,
"range": "all",
"c": "finance",
"page": page
}
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Cookie": "你的Cookie值" # 建议从浏览器复制
}
resp = requests.get(url, params=params, headers=headers)
# 这里需要用BeautifulSoup解析HTML
# 具体解析逻辑略，因为新浪经常改版
return articles_df4.2 社交媒体数据获取：Twitter与微博API
说完新闻，咱们聊聊社交媒体。为什么需要这个？因为很多市场情绪在新闻出来之前，就已经在社交媒体上发酵了。我记得2021年GameStop那波行情，Reddit上的帖子比任何新闻都快。
4.2.1 Twitter API实战
Twitter的API现在需要申请开发者账号，流程比以前严格了。但一旦拿到权限，用起来还是很顺手的。我个人习惯用Tweepy这个库，封装得不错。
import tweepy
import pandas as pd
# 配置你的API密钥
API_KEY = "你的API_KEY"
API_SECRET = "你的API_SECRET"
ACCESS_TOKEN = "你的ACCESS_TOKEN"
ACCESS_SECRET = "你的ACCESS_SECRET"
def fetch_tweets(keyword, count=100):
"""
抓取Twitter上关于某个关键词的推文
"""
auth = tweepy.OAuthHandler(API_KEY, API_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_SECRET)
api = tweepy.API(auth, wait_on_rate_limit=True)
tweets = []
try:
for tweet in tweepy.Cursor(
api.search_tweets,
q=keyword,
lang="en",
tweet_mode="extended"
).items(count):
tweets.append({
"id": tweet.id_str,
"text": tweet.full_text,
"created_at": tweet.created_at,
"user": tweet.user.screen_name,
"retweet_count": tweet.retweet_count,
"favorite_count": tweet.favorite_count
})
except tweepy.TweepError as e:
print(f"Twitter API错误: {e}")
return pd.DataFrame(tweets)
# 使用示例
df_tweets = fetch_tweets("$AAPL", count=50)
print(f"抓取了 {len(df_tweets)} 条推文")关键点： Twitter的免费API有速率限制，每15分钟只能请求450次。我建议你设置一个合理的抓取间隔，比如每次抓完休息15秒。另外，记得用wait_on_rate_limit=True，这样Tweepy会自动等待。
4.2.2 微博API实战
微博的API比Twitter复杂一些。它需要申请企业级开发者权限，个人开发者能拿到的接口有限。不过别担心，我有个替代方案——用模拟浏览器的方式抓取。
from selenium import webdriver
from selenium.webdriver.common.by import By
import time
import pandas as pd
def fetch_weibo_posts(keyword, max_pages=5):
"""
通过模拟浏览器抓取微博数据
"""
options = webdriver.ChromeOptions()
options.add_argument("--headless") # 无头模式
driver = webdriver.Chrome(options=options)
posts = []
try:
for page in range(1, max_pages + 1):
url = f"https://s.weibo.com/weibo?q={keyword}&page={page}"
driver.get(url)
time.sleep(3) # 等待页面加载
# 解析微博内容
items = driver.find_elements(By.CSS_SELECTOR, ".card-wrap")
for item in items:
try:
text = item.find_element(By.CSS_SELECTOR, ".txt").text
posts.append({
"text": text,
"page": page
})
except:
continue
finally:
driver.quit()
return pd.DataFrame(posts)
# 使用示例
df_weibo = fetch_weibo_posts("A股", max_pages=3)
print(f"抓取了 {len(df_weibo)} 条微博")4.3 数据清洗与存储
数据抓下来只是第一步。你想想看，原始数据里有多少噪音？HTML标签、特殊字符、重复内容...不处理干净，后面分析全是错的。
我一般会做这几步清洗：
- 去重： 同一篇新闻可能被多个源转载，用标题MD5去重
- 清洗文本： 去掉HTML标签、多余空格、特殊符号
- 标准化时间： 统一转为UTC时间戳
- 情感标注： 用预训练模型打上情感标签（后续章节会细讲）
def clean_news_data(df):
"""
清洗新闻数据
"""
import re
from hashlib import md5
# 去重
df["title_hash"] = df["title"].apply(lambda x: md5(x.encode()).hexdigest())
df = df.drop_duplicates(subset=["title_hash"])
# 清洗文本
def clean_text(text):
text = re.sub(r'<[^>]+>', '', text) # 去HTML标签
text = re.sub(r'\s+', ' ', text) # 合并空格
return text.strip()
df["clean_content"] = df["content"].apply(clean_text)
# 标准化时间
df["timestamp"] = pd.to_datetime(df["date"])
return df
# 存储到CSV
df_clean = clean_news_data(df_news)
df_clean.to_csv("news_data.csv", index=False)4.4 本章知识体系
说了这么多，我画了一张图帮你理清思路。这张图展示了从数据源到最终存储的完整流程。
这张图把整个流程分成了四层：数据源、抓取方式、数据清洗、存储。每一层都有对应的技术选型。我个人建议，优先用API接口，实在不行再上模拟浏览器。毕竟API稳定，不容易被封。
好了，这一章的内容就到这里。数据源的问题解决了，后面咱们就可以开始做真正的情绪分析了。记住，数据质量决定了分析的上限。把基础打牢，后面的路才好走。