市场情绪分析是量化交易中的重要分支，通过分析各类文本数据（如新闻、社交媒体、研报等）来捕捉市场参与者的情绪变化，进而预测市场走势。本项目旨在帮助学习者构建一个完整的基于市场情绪分析的量化交易策略，覆盖数据采集、文本处理、情绪分析、策略设计与回测优化等环节。
完成本项目后，学习者将能够：
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
import os
from dotenv import load_dotenv
load_dotenv() # 加载环境变量
class DataCollector:
def __init__(self):
# 各API密钥
self.news_api_key = os.getenv("NEWS_API_KEY")
self.twitter_bearer_token = os.getenv("TWITTER_BEARER_TOKEN")
self.reddit_client_id = os.getenv("REDDIT_CLIENT_ID")
self.reddit_client_secret = os.getenv("REDDIT_CLIENT_SECRET")
def get_financial_news(self, keywords, start_date, end_date, max_results=100):
"""
从News API获取金融新闻
参数:
keywords (str): 搜索关键词，如"Bitcoin,Ethereum"
start_date (str): 开始日期，格式 YYYY-MM-DD
end_date (str): 结束日期，格式 YYYY-MM-DD
max_results (int): 最大返回结果数
返回:
pd.DataFrame: 包含新闻数据的DataFrame
"""
url = "https://newsapi.org/v2/everything"
params = {
'q': keywords,
'from': start_date,
'to': end_date,
'language': 'en',
'sortBy': 'publishedAt',
'pageSize': min(max_results, 100), # API限制
'apiKey': self.news_api_key
}
response = requests.get(url, params=params)
if response.status_code != 200:
print(f"Error: {response.status_code}, {response.text}")
return pd.DataFrame()
articles = response.json().get('articles', [])
# 创建DataFrame
if not articles:
return pd.DataFrame()
df = pd.DataFrame(articles)
df['publishedAt'] = pd.to_datetime(df['publishedAt'])
# 提取重要字段
df = df[['title', 'description', 'content', 'publishedAt', 'source', 'url']]
return df
def get_twitter_data(self, query, start_time, end_time, max_results=100):
"""
从Twitter API v2获取推文数据
参数:
query (str): 搜索查询，如"#Bitcoin"
start_time (str): 开始时间，格式 YYYY-MM-DDTHH:MM:SSZ
end_time (str): 结束时间，格式 YYYY-MM-DDTHH:MM:SSZ
max_results (int): 最大返回结果数
返回:
pd.DataFrame: 包含推文数据的DataFrame
"""
url = "https://api.twitter.com/2/tweets/search/recent"
headers = {
"Authorization": f"Bearer {self.twitter_bearer_token}"
}
params = {
'query': query,
'start_time': start_time,
'end_time': end_time,
'max_results': min(max_results, 100), # API限制
'tweet.fields': 'created_at,public_metrics,lang'
}
response = requests.get(url, headers=headers, params=params)
if response.status_code != 200:
print(f"Error: {response.status_code}, {response.text}")
return pd.DataFrame()
tweets = response.json().get('data', [])
# 创建DataFrame
if not tweets:
return pd.DataFrame()
df = pd.DataFrame(tweets)
df['created_at'] = pd.to_datetime(df['created_at'])
# 展开public_metrics
metrics_df = pd.json_normalize(df['public_metrics'])
df = pd.concat([df.drop('public_metrics', axis=1), metrics_df], axis=1)
return df
def get_reddit_data(self, subreddit, time_filter='week', limit=100):
"""
从Reddit API获取帖子数据
参数:
subreddit (str): 子版块名称，如"Bitcoin"
time_filter (str): 时间过滤器，如'day', 'week', 'month'
limit (int): 最大返回结果数
返回:
pd.DataFrame: 包含Reddit帖子数据的DataFrame
"""
# 获取认证token
auth = requests.auth.HTTPBasicAuth(self.reddit_client_id, self.reddit_client_secret)
data = {
'grant_type': 'password',
'username': os.getenv("REDDIT_USERNAME"),
'password': os.getenv("REDDIT_PASSWORD")
}
headers = {'User-Agent': 'FinanceApp/0.1'}
response = requests.post(
'https://www.reddit.com/api/v1/access_token',
auth=auth,
data=data,
headers=headers
)
if response.status_code != 200:
print(f"Authentication Error: {response.status_code}, {response.text}")
return pd.DataFrame()
token = response.json()['access_token']
headers['Authorization'] = f'bearer {token}'
# 获取帖子数据
url = f"https://oauth.reddit.com/r/{subreddit}/top"
params = {
't': time_filter,
'limit': min(limit, 100) # API限制
}
response = requests.get(url, headers=headers, params=params)
if response.status_code != 200:
print(f"Error: {response.status_code}, {response.text}")
return pd.DataFrame()
posts = response.json()['data']['children']
# 创建DataFrame
if not posts:
return pd.DataFrame()
data = []
for post in posts:
post_data = post['data']
data.append({
'title': post_data.get('title'),
'selftext': post_data.get('selftext'),
'created_utc': datetime.fromtimestamp(post_data.get('created_utc')),
'score': post_data.get('score'),
'num_comments': post_data.get('num_comments'),
'upvote_ratio': post_data.get('upvote_ratio'),
'permalink': post_data.get('permalink')
})
df = pd.DataFrame(data)
return df
def save_data(self, df, filename):
"""保存数据到CSV文件"""
df.to_csv(filename, index=False)
print(f"Data saved to {filename}")
# 使用示例
if __name__ == "__main__":
collector = DataCollector()
# 获取比特币相关新闻
end_date = datetime.now().strftime('%Y-%m-%d')
start_date = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d')
news_df = collector.get_financial_news("Bitcoin OR Cryptocurrency", start_date, end_date)
collector.save_data(news_df, "bitcoin_news.csv")
# 注意：以下代码需要有对应的API密钥才能执行
# Twitter数据示例
# twitter_df = collector.get_twitter_data("#Bitcoin",
# (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%dT%H:%M:%SZ'),
# datetime.now().strftime('%Y-%m-%dT%H:%M:%SZ'))
# collector.save_data(twitter_df, "bitcoin_tweets.csv")
# Reddit数据示例
# reddit_df = collector.get_reddit_data("Bitcoin", 'week')
# collector.save_data(reddit_df, "bitcoin_reddit.csv")
import pandas as pd
import numpy as np
import re
import nltk
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize
from nltk.stem import WordNetLemmatizer
import spacy
import textblob
from textblob import TextBlob
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch
# 下载必要的NLTK数据
nltk.download('punkt')
nltk.download('stopwords')
nltk.download('wordnet')
class TextProcessor:
def __init__(self, use_spacy=False, use_transformers=False):
"""
初始化文本处理器
参数:
use_spacy (bool): 是否使用spaCy进行高级NLP处理
use_transformers (bool): 是否使用Transformers模型进行情感分析
"""
self.lemmatizer = WordNetLemmatizer()
self.stop_words = set(stopwords.words('english'))
self.vader_analyzer = SentimentIntensityAnalyzer()
# 添加金融领域的特定停用词
finance_stopwords = {'nasdaq', 'nyse', 'market', 'stock', 'share', 'price',
'index', 'exchange', 'trading', 'trader', 'investor'}
self.stop_words.update(finance_stopwords)
# 加载spaCy模型（可选）
self.use_spacy = use_spacy
if use_spacy:
try:
self.nlp = spacy.load("en_core_web_sm")
except:
print("spaCy模型未找到。请运行: python -m spacy download en_core_web_sm")
self.use_spacy = False
# 加载transformers模型（可选）
self.use_transformers = use_transformers
if use_transformers:
try:
model_name = "ProsusAI/finbert"
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
self.finbert_pipeline = pipeline("sentiment-analysis", model=self.model, tokenizer=self.tokenizer)
except:
print("无法加载FinBERT模型。请检查您的互联网连接和transformers库安装。")
self.use_transformers = False
def clean_text(self, text):
"""
清洗文本数据
参数:
text (str): 输入文本
返回:
str: 清洗后的文本
"""
if not isinstance(text, str):
return ""
# 转换为小写
text = text.lower()
# 移除URL
text = re.sub(r'http\S+', '', text)
# 移除@用户名
text = re.sub(r'@\w+', '', text)
# 移除标签符号
text = re.sub(r'#', '', text)
# 移除特殊字符和数字
text = re.sub(r'[^a-zA-Z\s]', '', text)
# 移除多余空格
text = re.sub(r'\s+', ' ', text).strip()
return text
def tokenize_text(self, text):
"""
文本分词
参数:
text (str): 输入文本
返回:
list: 分词后的token列表
"""
tokens = word_tokenize(text)
# 移除停用词
tokens = [token for token in tokens if token not in self.stop_words]
# 词形还原
tokens = [self.lemmatizer.lemmatize(token) for token in tokens]
return tokens
def extract_entities(self, text):
"""
使用spaCy提取命名实体
参数:
text (str): 输入文本
返回:
list: 提取的实体列表
"""
if not self.use_spacy:
return []
doc = self.nlp(text)
entities = [(ent.text, ent.label_) for ent in doc.ents]
return entities
def get_textblob_sentiment(self, text):
"""
使用TextBlob计算情感分数
参数:
text (str): 输入文本
返回:
tuple: (极性, 主观性)
"""
blob = TextBlob(text)
return blob.sentiment.polarity, blob.sentiment.subjectivity
def get_vader_sentiment(self, text):
"""
使用VADER计算情感分数
参数:
text (str): 输入文本
返回:
dict: 包含pos, neg, neu和compound分数的字典
"""
return self.vader_analyzer.polarity_scores(text)
def get_finbert_sentiment(self, text):
"""
使用FinBERT计算金融特定情感
参数:
text (str): 输入文本
返回:
dict: 情感分析结果
"""
if not self.use_transformers:
return {'label': 'neutral', 'score': 0.5}
# 限制文本长度
max_length = 512
if len(text) > max_length:
text = text[:max_length]
try:
result = self.finbert_pipeline(text)[0]
return result
except Exception as e:
print(f"FinBERT分析错误: {e}")
return {'label': 'neutral', 'score': 0.5}
def process_dataframe(self, df, text_column, new_column_prefix='processed'):
"""
处理DataFrame中的文本列
参数:
df (pd.DataFrame): 输入数据框
text_column (str): 包含文本的列名
new_column_prefix (str): 新列的前缀
返回:
pd.DataFrame: 处理后的数据框
"""
# 创建DataFrame的副本
result_df = df.copy()
# 清洗文本
result_df[f'{new_column_prefix}_text'] = result_df[text_column].apply(self.clean_text)
# 分词并计算token长度
result_df[f'{new_column_prefix}_tokens'] = result_df[f'{new_column_prefix}_text'].apply(self.tokenize_text)
result_df[f'{new_column_prefix}_token_count'] = result_df[f'{new_column_prefix}_tokens'].apply(len)
# TextBlob情感分析
sentiment_results = result_df[f'{new_column_prefix}_text'].apply(self.get_textblob_sentiment)
result_df[f'{new_column_prefix}_polarity'] = sentiment_results.apply(lambda x: x[0])
result_df[f'{new_column_prefix}_subjectivity'] = sentiment_results.apply(lambda x: x[1])
# VADER情感分析
vader_results = result_df[f'{new_column_prefix}_text'].apply(self.get_vader_sentiment)
result_df[f'{new_column_prefix}_vader_neg'] = vader_results.apply(lambda x: x['neg'])
result_df[f'{new_column_prefix}_vader_neu'] = vader_results.apply(lambda x: x['neu'])
result_df[f'{new_column_prefix}_vader_pos'] = vader_results.apply(lambda x: x['pos'])
result_df[f'{new_column_prefix}_vader_compound'] = vader_results.apply(lambda x: x['compound'])
# 如果启用，添加FinBERT情感分析
if self.use_transformers:
finbert_results = result_df[f'{new_column_prefix}_text'].apply(self.get_finbert_sentiment)
result_df[f'{new_column_prefix}_finbert_label'] = finbert_results.apply(lambda x: x['label'])
result_df[f'{new_column_prefix}_finbert_score'] = finbert_results.apply(lambda x: x['score'])
# 如果启用，添加命名实体识别
if self.use_spacy:
result_df[f'{new_column_prefix}_entities'] = result_df[f'{new_column_prefix}_text'].apply(self.extract_entities)
return result_df
# 使用示例
if __name__ == "__main__":
# 创建示例数据
data = {
'date': pd.date_range(start='2023-01-01', periods=3),
'headline': [
"Bitcoin surges to new all-time high as institutional investors pile in",
"Markets crash as inflation fears grow and central bank hints at rate hikes",
"Tech stocks rally despite mixed earnings reports from FAANG companies"
]
}
df = pd.DataFrame(data)
# 初始化文本处理器（简单版本，不使用spaCy和transformers以避免依赖问题）
processor = TextProcessor(use_spacy=False, use_transformers=False)
# 处理文本
processed_df = processor.process_dataframe(df, 'headline')
# 显示结果
print(processed_df.head())