4、数据采集实战：使用Tweepy抓取Twitter数据、处理API限流、数据存储为CSV/Parquet、增量更新策略
好，咱们直接进入正题。数据采集这件事，说白了就是量化策略的「水源」。水源不干净，后面所有模型都是白搭。我见过太多人把精力全花在策略优化上，结果数据源一塌糊涂——嗯，我自己早期也踩过这个坑。
4.1 环境准备与Tweepy基础
先搞定环境。我个人习惯用Python 3.9+，Tweepy 4.x版本。安装就一行命令：
pip install tweepy pandas pyarrow然后你需要去Twitter Developer Portal申请API Key。这里有个坑——记得申请Elevated Access，否则你只能拿到极其有限的采样数据。我当初用Basic权限跑了三天，发现数据量根本不够做策略回测，白白浪费了时间。
认证代码长这样：
import tweepy
client = tweepy.Client(
bearer_token='你的Bearer Token',
consumer_key='你的API Key',
consumer_secret='你的API Secret',
access_token='你的Access Token',
access_token_secret='你的Access Token Secret'
)4.2 抓取推文数据——从关键词到时间线
实战中最常用的场景是按关键词搜索。比如我们要抓取「比特币」相关的推文：
query = '比特币 lang:en -is:retweet'
tweets = client.search_recent_tweets(
query=query,
max_results=100,
tweet_fields=['created_at', 'public_metrics', 'author_id']
)
for tweet in tweets.data:
print(tweet.text[:50])这里我用了 -is:retweet 过滤掉转发，因为转发通常不包含新的情绪信息。你想想看，一条推文被转一万次，情绪还是原来那条，对策略来说就是重复数据。
另外，tweet_fields 这个参数很关键。我建议至少抓取：
created_at—— 时间戳，做时序分析必备public_metrics—— 包含like_count、retweet_count、reply_countauthor_id—— 用于后续分析KOL影响力
4.3 处理API限流——别让你的脚本突然挂掉
Twitter API有严格的速率限制。免费账号每15分钟只能请求450次。如果你不管这个限制，脚本跑着跑着就报 429 Too Many Requests。
为什么会这样？因为Twitter要保护它的服务器资源。我早期做项目时，脚本半夜跑崩了，第二天起来发现数据只抓了一半，回测结果完全不能用。
解决方案是加一个「优雅的限流处理」：
import time
from tweepy.errors import TooManyRequests
def fetch_with_retry(client, query, max_retries=5):
for attempt in range(max_retries):
try:
return client.search_recent_tweets(query=query, max_results=100)
except TooManyRequests:
wait_time = 15 * 60 # 15分钟
print(f'限流了，等待{wait_time}秒...')
time.sleep(wait_time)
raise Exception('重试次数耗尽')更高级的做法是动态计算等待时间。Twitter会在响应头里告诉你 x-rate-limit-reset，你可以解析这个时间戳，精确计算需要等多久。
4.4 数据存储——CSV还是Parquet？
数据抓下来得存起来。这里我给出我的选择逻辑：
| 存储格式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| CSV | 小规模数据、人工查看 | 通用、可读性强 | 占用空间大、读写慢 |
| Parquet | 大规模数据、机器学习 | 压缩率高、读写快 | 二进制格式、不易直接查看 |
我个人习惯是：原始数据存Parquet，抽样数据存CSV。这样既保留了完整数据用于建模，又方便日常快速查看。
存Parquet的代码：
import pandas as pd
df = pd.DataFrame([tweet.data for tweet in tweets.data])
df.to_parquet('btc_tweets.parquet', engine='pyarrow')存CSV就更简单了：
df.to_csv('btc_tweets_sample.csv', index=False)4.5 增量更新策略——别每次都全量抓
全量抓取是新手最容易犯的错误。每次跑脚本都把历史数据重新抓一遍，不仅浪费API额度，还容易触发限流。
正确的做法是增量更新。思路很简单：
- 记录上次抓取的最新推文ID（称为
since_id） - 下次只抓取比这个ID更新的推文
- 合并到已有数据中
代码实现：
import json
# 读取上次的since_id
try:
with open('since_id.json', 'r') as f:
since_id = json.load(f)['since_id']
except FileNotFoundError:
since_id = None
# 增量抓取
tweets = client.search_recent_tweets(
query=query,
max_results=100,
since_id=since_id
)
# 更新since_id
if tweets.data:
new_since_id = tweets.data[0].id
with open('since_id.json', 'w') as f:
json.dump({'since_id': new_since_id}, f)这里有个细节：since_id 存的是推文ID，不是时间戳。因为推文ID是递增的，用ID比用时间更精确，不会出现时区问题。
4.6 知识体系总览
下面这张图概括了本章的核心逻辑，从数据源到存储再到增量更新，一条线串下来：
这张图把整个流程串起来了。你从Twitter API拿到原始数据，经过关键词筛选和限流保护，存到CSV或Parquet里，最后用增量策略持续更新。环环相扣，少一步都不行。
4.7 实战中的几个坑
- 数据去重：Twitter API偶尔会返回重复数据。我建议在入库前用
tweet.id做一次去重，避免后续分析偏差。 - 文本清洗：推文里全是URL、@提及、表情符号。我习惯用正则把URL和@去掉，但保留表情符号——因为表情符号本身就有情绪价值。
- 存储路径：按日期分文件夹存，比如
data/2024/01/btc_tweets.parquet。这样回溯数据时，直接按日期范围加载，效率高很多。