mirror of
https://github.com/B-Wear/QuantumEdge.git
synced 2026-07-27 15:37:46 +00:00
B-Wear
267 lines
9.6 KiB
Python
267 lines
9.6 KiB
Python
import numpy as np
|
|
import pandas as pd
|
|
from typing import Dict, List, Optional
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
import requests
|
|
from textblob import TextBlob
|
|
import tweepy
|
|
from newsapi import NewsApiClient
|
|
import yfinance as yf
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import json
|
|
import os
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class SentimentAnalyzer:
|
|
def __init__(self, config: Dict):
|
|
self.config = config
|
|
self.news_api = NewsApiClient(api_key=config['news_api_key'])
|
|
self.twitter_auth = tweepy.OAuthHandler(
|
|
config['twitter_api_key'],
|
|
config['twitter_api_secret']
|
|
)
|
|
self.twitter_auth.set_access_token(
|
|
config['twitter_access_token'],
|
|
config['twitter_access_token_secret']
|
|
)
|
|
self.twitter_api = tweepy.API(self.twitter_auth)
|
|
|
|
# Initialize sentiment cache
|
|
self.sentiment_cache = {}
|
|
self.cache_duration = timedelta(hours=1)
|
|
|
|
def analyze_text(self, text: str) -> float:
|
|
"""
|
|
Analyze sentiment of a single text using TextBlob
|
|
"""
|
|
try:
|
|
analysis = TextBlob(text)
|
|
# Normalize sentiment score to [-1, 1]
|
|
return analysis.sentiment.polarity
|
|
except Exception as e:
|
|
logger.error(f"Error analyzing text: {str(e)}")
|
|
return 0.0
|
|
|
|
def get_news_sentiment(self, symbol: str) -> Optional[float]:
|
|
"""
|
|
Get sentiment from news articles
|
|
"""
|
|
try:
|
|
# Check cache first
|
|
cache_key = f"news_{symbol}"
|
|
if cache_key in self.sentiment_cache:
|
|
cached_data = self.sentiment_cache[cache_key]
|
|
if datetime.now() - cached_data['timestamp'] < self.cache_duration:
|
|
return cached_data['sentiment']
|
|
|
|
# Get news articles
|
|
news = self.news_api.get_everything(
|
|
q=symbol,
|
|
language='en',
|
|
from_param=(datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d'),
|
|
sort_by='relevancy'
|
|
)
|
|
|
|
if not news['articles']:
|
|
return None
|
|
|
|
# Analyze sentiment of each article
|
|
sentiments = []
|
|
for article in news['articles']:
|
|
title_sentiment = self.analyze_text(article['title'])
|
|
if article['description']:
|
|
desc_sentiment = self.analyze_text(article['description'])
|
|
sentiments.append((title_sentiment + desc_sentiment) / 2)
|
|
else:
|
|
sentiments.append(title_sentiment)
|
|
|
|
# Calculate weighted average sentiment
|
|
avg_sentiment = np.mean(sentiments)
|
|
|
|
# Cache the result
|
|
self.sentiment_cache[cache_key] = {
|
|
'sentiment': avg_sentiment,
|
|
'timestamp': datetime.now()
|
|
}
|
|
|
|
return avg_sentiment
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting news sentiment: {str(e)}")
|
|
return None
|
|
|
|
def get_twitter_sentiment(self, symbol: str) -> Optional[float]:
|
|
"""
|
|
Get sentiment from Twitter
|
|
"""
|
|
try:
|
|
# Check cache first
|
|
cache_key = f"twitter_{symbol}"
|
|
if cache_key in self.sentiment_cache:
|
|
cached_data = self.sentiment_cache[cache_key]
|
|
if datetime.now() - cached_data['timestamp'] < self.cache_duration:
|
|
return cached_data['sentiment']
|
|
|
|
# Get tweets
|
|
tweets = self.twitter_api.search_tweets(
|
|
q=f"${symbol}",
|
|
lang="en",
|
|
count=100
|
|
)
|
|
|
|
if not tweets:
|
|
return None
|
|
|
|
# Analyze sentiment of each tweet
|
|
sentiments = []
|
|
for tweet in tweets:
|
|
sentiment = self.analyze_text(tweet.text)
|
|
sentiments.append(sentiment)
|
|
|
|
# Calculate weighted average sentiment
|
|
avg_sentiment = np.mean(sentiments)
|
|
|
|
# Cache the result
|
|
self.sentiment_cache[cache_key] = {
|
|
'sentiment': avg_sentiment,
|
|
'timestamp': datetime.now()
|
|
}
|
|
|
|
return avg_sentiment
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting Twitter sentiment: {str(e)}")
|
|
return None
|
|
|
|
def get_market_sentiment(self, symbol: str) -> Optional[float]:
|
|
"""
|
|
Get market sentiment indicators
|
|
"""
|
|
try:
|
|
# Check cache first
|
|
cache_key = f"market_{symbol}"
|
|
if cache_key in self.sentiment_cache:
|
|
cached_data = self.sentiment_cache[cache_key]
|
|
if datetime.now() - cached_data['timestamp'] < self.cache_duration:
|
|
return cached_data['sentiment']
|
|
|
|
# Get market data
|
|
ticker = yf.Ticker(symbol)
|
|
info = ticker.info
|
|
|
|
# Calculate various sentiment indicators
|
|
sentiment_indicators = []
|
|
|
|
# RSI sentiment
|
|
if 'RSI' in info:
|
|
rsi = info['RSI']
|
|
rsi_sentiment = (rsi - 50) / 50 # Normalize to [-1, 1]
|
|
sentiment_indicators.append(rsi_sentiment)
|
|
|
|
# Volume sentiment
|
|
if 'volume' in info and 'averageVolume' in info:
|
|
volume_ratio = info['volume'] / info['averageVolume']
|
|
volume_sentiment = (volume_ratio - 1) / volume_ratio # Normalize to [-1, 1]
|
|
sentiment_indicators.append(volume_sentiment)
|
|
|
|
# Price momentum sentiment
|
|
if 'regularMarketChangePercent' in info:
|
|
momentum_sentiment = info['regularMarketChangePercent'] / 100
|
|
sentiment_indicators.append(momentum_sentiment)
|
|
|
|
if not sentiment_indicators:
|
|
return None
|
|
|
|
# Calculate weighted average sentiment
|
|
avg_sentiment = np.mean(sentiment_indicators)
|
|
|
|
# Cache the result
|
|
self.sentiment_cache[cache_key] = {
|
|
'sentiment': avg_sentiment,
|
|
'timestamp': datetime.now()
|
|
}
|
|
|
|
return avg_sentiment
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting market sentiment: {str(e)}")
|
|
return None
|
|
|
|
def get_combined_sentiment(self, symbol: str) -> Optional[float]:
|
|
"""
|
|
Get combined sentiment from all sources
|
|
"""
|
|
try:
|
|
# Check cache first
|
|
cache_key = f"combined_{symbol}"
|
|
if cache_key in self.sentiment_cache:
|
|
cached_data = self.sentiment_cache[cache_key]
|
|
if datetime.now() - cached_data['timestamp'] < self.cache_duration:
|
|
return cached_data['sentiment']
|
|
|
|
# Get sentiment from all sources
|
|
sentiments = []
|
|
weights = []
|
|
|
|
# News sentiment
|
|
news_sentiment = self.get_news_sentiment(symbol)
|
|
if news_sentiment is not None:
|
|
sentiments.append(news_sentiment)
|
|
weights.append(self.config['sentiment_weights']['news'])
|
|
|
|
# Twitter sentiment
|
|
twitter_sentiment = self.get_twitter_sentiment(symbol)
|
|
if twitter_sentiment is not None:
|
|
sentiments.append(twitter_sentiment)
|
|
weights.append(self.config['sentiment_weights']['twitter'])
|
|
|
|
# Market sentiment
|
|
market_sentiment = self.get_market_sentiment(symbol)
|
|
if market_sentiment is not None:
|
|
sentiments.append(market_sentiment)
|
|
weights.append(self.config['sentiment_weights']['market'])
|
|
|
|
if not sentiments:
|
|
return None
|
|
|
|
# Calculate weighted average sentiment
|
|
avg_sentiment = np.average(sentiments, weights=weights)
|
|
|
|
# Cache the result
|
|
self.sentiment_cache[cache_key] = {
|
|
'sentiment': avg_sentiment,
|
|
'timestamp': datetime.now()
|
|
}
|
|
|
|
return avg_sentiment
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting combined sentiment: {str(e)}")
|
|
return None
|
|
|
|
def clear_cache(self):
|
|
"""
|
|
Clear the sentiment cache
|
|
"""
|
|
self.sentiment_cache.clear()
|
|
|
|
def get_sentiment_signal(self, symbol: str) -> int:
|
|
"""
|
|
Convert sentiment to trading signal
|
|
"""
|
|
sentiment = self.get_combined_sentiment(symbol)
|
|
if sentiment is None:
|
|
return 0
|
|
|
|
# Define sentiment thresholds
|
|
positive_threshold = self.config['sentiment_thresholds']['positive']
|
|
negative_threshold = self.config['sentiment_thresholds']['negative']
|
|
|
|
if sentiment > positive_threshold:
|
|
return 1 # Buy signal
|
|
elif sentiment < negative_threshold:
|
|
return -1 # Sell signal
|
|
else:
|
|
return 0 # Neutral signal |