You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
544 lines
22 KiB
Python
544 lines
22 KiB
Python
"""
|
|
News and Social Media Data Aggregation Service
|
|
|
|
Aggregates news and social media data from multiple sources for sentiment analysis:
|
|
- Yahoo Finance news (via yfinance_plus)
|
|
- NewsAPI
|
|
- Reddit API
|
|
"""
|
|
|
|
import asyncio
|
|
import aiohttp
|
|
import logging
|
|
|
|
from app.core.http_client import get_http_session
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional, Any, Union
|
|
from dataclasses import dataclass
|
|
import json
|
|
import re
|
|
import time
|
|
|
|
# Import yfinance_plus for news data
|
|
import sys
|
|
import os
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), '../../yfinance_plus'))
|
|
from yfinance_plus import Ticker
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class NewsArticle:
|
|
"""Standardized news article data structure"""
|
|
title: str
|
|
summary: Optional[str]
|
|
content: Optional[str]
|
|
url: str
|
|
source: str
|
|
published_at: datetime
|
|
author: Optional[str] = None
|
|
relevance_score: Optional[float] = None
|
|
image_url: Optional[str] = None
|
|
tags: List[str] = None
|
|
|
|
def __post_init__(self):
|
|
if self.tags is None:
|
|
self.tags = []
|
|
|
|
def to_dict(self) -> Dict:
|
|
"""Convert to dictionary for API response"""
|
|
return {
|
|
"title": self.title,
|
|
"summary": self.summary,
|
|
"content": self.content,
|
|
"url": self.url,
|
|
"source": self.source,
|
|
"published_at": self.published_at.isoformat() if self.published_at else None,
|
|
"author": self.author,
|
|
"relevance_score": self.relevance_score,
|
|
"image_url": self.image_url,
|
|
"tags": self.tags
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class SocialPost:
|
|
"""Standardized social media post data structure"""
|
|
title: str
|
|
content: str
|
|
url: str
|
|
platform: str
|
|
author: str
|
|
published_at: datetime
|
|
score: Optional[int] = None
|
|
comments_count: Optional[int] = None
|
|
upvotes: Optional[int] = None
|
|
downvotes: Optional[int] = None
|
|
subreddit: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict:
|
|
"""Convert to dictionary for API response"""
|
|
return {
|
|
"title": self.title,
|
|
"content": self.content,
|
|
"url": self.url,
|
|
"platform": self.platform,
|
|
"author": self.author,
|
|
"published_at": self.published_at.isoformat() if self.published_at else None,
|
|
"score": self.score,
|
|
"comments_count": self.comments_count,
|
|
"upvotes": self.upvotes,
|
|
"downvotes": self.downvotes,
|
|
"subreddit": self.subreddit
|
|
}
|
|
|
|
|
|
class NewsAPIError(Exception):
|
|
"""News API related errors"""
|
|
pass
|
|
|
|
|
|
class RedditAPIError(Exception):
|
|
"""Reddit API related errors"""
|
|
pass
|
|
|
|
|
|
class NewsSocialService:
|
|
"""Service for aggregating news and social media data"""
|
|
|
|
def __init__(self):
|
|
# API credentials
|
|
self.newsapi_key = "04169755c1a34a4593316855c56adc3f"
|
|
self.reddit_client_id = "vVHvj_0Yj9wtmEjQoPYIIg"
|
|
self.reddit_client_secret = "jFRqcoryQFDIJbHeIIn93h18hUGFVg"
|
|
|
|
# Reddit access token (will be obtained dynamically)
|
|
self._reddit_token = None
|
|
self._reddit_token_expiry = None
|
|
|
|
# Rate limiting
|
|
self._last_newsapi_request = 0
|
|
self._last_reddit_request = 0
|
|
self._newsapi_rate_limit = 1.0 # 1 second between requests
|
|
self._reddit_rate_limit = 1.0 # 1 second between requests
|
|
|
|
async def get_ticker_news_and_social(
|
|
self,
|
|
ticker: str,
|
|
days_back: int = 7,
|
|
max_articles: int = 20,
|
|
max_social_posts: int = 15,
|
|
include_social: bool = True
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Get comprehensive news and social media data for a ticker
|
|
|
|
Args:
|
|
ticker: Stock ticker symbol (e.g., "AAPL", "TSLA")
|
|
days_back: Number of days to look back for articles
|
|
max_articles: Maximum number of news articles to return
|
|
max_social_posts: Maximum number of social media posts to return
|
|
include_social: Whether to include social media data
|
|
|
|
Returns:
|
|
Dictionary with news and social media data
|
|
"""
|
|
try:
|
|
# Run all data collection in parallel
|
|
tasks = []
|
|
|
|
# Yahoo Finance news
|
|
tasks.append(self._get_yahoo_news(ticker, max_articles // 3))
|
|
|
|
# NewsAPI
|
|
tasks.append(self._get_newsapi_articles(ticker, days_back, max_articles // 3))
|
|
|
|
# Reddit data (if enabled)
|
|
if include_social:
|
|
tasks.append(self._get_reddit_posts(ticker, days_back, max_social_posts))
|
|
else:
|
|
tasks.append(asyncio.create_task(self._empty_social_data()))
|
|
|
|
# Execute all tasks in parallel
|
|
yahoo_news, newsapi_articles, reddit_posts = await asyncio.gather(
|
|
*tasks, return_exceptions=True
|
|
)
|
|
|
|
# Handle exceptions
|
|
if isinstance(yahoo_news, Exception):
|
|
logger.error(f"Yahoo Finance news error: {yahoo_news}")
|
|
yahoo_news = []
|
|
|
|
if isinstance(newsapi_articles, Exception):
|
|
logger.error(f"NewsAPI error: {newsapi_articles}")
|
|
newsapi_articles = []
|
|
|
|
if isinstance(reddit_posts, Exception):
|
|
logger.error(f"Reddit API error: {reddit_posts}")
|
|
reddit_posts = []
|
|
|
|
# Combine and deduplicate articles
|
|
all_articles = []
|
|
all_articles.extend(yahoo_news)
|
|
all_articles.extend(newsapi_articles)
|
|
|
|
# Sort articles by published date (newest first)
|
|
all_articles.sort(key=lambda x: x.published_at or datetime.min, reverse=True)
|
|
|
|
# Limit total articles
|
|
if len(all_articles) > max_articles:
|
|
all_articles = all_articles[:max_articles]
|
|
|
|
# Sort social posts by score/engagement (if available)
|
|
if reddit_posts:
|
|
reddit_posts.sort(key=lambda x: x.score or 0, reverse=True)
|
|
if len(reddit_posts) > max_social_posts:
|
|
reddit_posts = reddit_posts[:max_social_posts]
|
|
|
|
# Compile response
|
|
response = {
|
|
"ticker": ticker.upper(),
|
|
"retrieved_at": datetime.now().isoformat(),
|
|
"news": {
|
|
"total_articles": len(all_articles),
|
|
"sources": {
|
|
"yahoo_finance": len([a for a in all_articles if a.source == "Yahoo Finance"]),
|
|
"newsapi": len([a for a in all_articles if a.source == "NewsAPI"]),
|
|
},
|
|
"articles": [article.to_dict() for article in all_articles]
|
|
},
|
|
"social_media": {
|
|
"total_posts": len(reddit_posts),
|
|
"platforms": {
|
|
"reddit": len(reddit_posts)
|
|
},
|
|
"posts": [post.to_dict() for post in reddit_posts] if include_social else []
|
|
},
|
|
"summary": {
|
|
"total_items": len(all_articles) + len(reddit_posts),
|
|
"time_range_days": days_back,
|
|
"oldest_item": min([a.published_at for a in all_articles + reddit_posts if a.published_at] or [datetime.now()]).isoformat() if all_articles + reddit_posts else datetime.now().isoformat(),
|
|
"newest_item": max([a.published_at for a in all_articles + reddit_posts if a.published_at] or [datetime.now()]).isoformat() if all_articles + reddit_posts else datetime.now().isoformat(),
|
|
}
|
|
}
|
|
|
|
return response
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting news and social data for {ticker}: {e}")
|
|
raise
|
|
|
|
async def _empty_social_data(self) -> List[SocialPost]:
|
|
"""Return empty social data when social media is disabled"""
|
|
return []
|
|
|
|
async def _get_yahoo_news(self, ticker: str, max_articles: int) -> List[NewsArticle]:
|
|
"""Get news from Yahoo Finance via yfinance_plus"""
|
|
try:
|
|
logger.info(f"Fetching Yahoo Finance news for {ticker}")
|
|
|
|
# Use yfinance_plus to get news data
|
|
ticker_obj = Ticker(ticker.upper())
|
|
news_data = ticker_obj.news
|
|
|
|
articles = []
|
|
if news_data:
|
|
for item in news_data[:max_articles]:
|
|
try:
|
|
# Parse new Yahoo Finance news format from yfinance_plus
|
|
content = item.get('content', {})
|
|
|
|
# Extract publication date
|
|
published_at = None
|
|
if 'pubDate' in content:
|
|
# Parse ISO format: 2025-06-09T20:06:19Z
|
|
pub_date_str = content['pubDate'].replace('Z', '+00:00')
|
|
published_at = datetime.fromisoformat(pub_date_str).replace(tzinfo=None)
|
|
elif 'displayTime' in content:
|
|
# Parse ISO format: 2025-08-10T16:27:54Z
|
|
display_time_str = content['displayTime'].replace('Z', '+00:00')
|
|
published_at = datetime.fromisoformat(display_time_str).replace(tzinfo=None)
|
|
else:
|
|
published_at = datetime.now()
|
|
|
|
# Extract thumbnail URL
|
|
image_url = None
|
|
thumbnail = content.get('thumbnail', {})
|
|
if thumbnail and 'resolutions' in thumbnail:
|
|
resolutions = thumbnail['resolutions']
|
|
if resolutions and len(resolutions) > 0:
|
|
# Use the largest resolution (last one)
|
|
image_url = resolutions[-1].get('url')
|
|
|
|
# Extract URL
|
|
url = ''
|
|
if 'canonicalUrl' in content:
|
|
url = content['canonicalUrl'].get('url', '')
|
|
elif 'clickThroughUrl' in content:
|
|
url = content['clickThroughUrl'].get('url', '')
|
|
|
|
# Extract provider
|
|
provider = content.get('provider', {})
|
|
author = provider.get('displayName', 'Yahoo Finance')
|
|
|
|
article = NewsArticle(
|
|
title=content.get('title', ''),
|
|
summary=content.get('description', '') or content.get('summary', ''),
|
|
content=None, # Yahoo Finance doesn't provide full content
|
|
url=url,
|
|
source="Yahoo Finance",
|
|
published_at=published_at,
|
|
author=author,
|
|
image_url=image_url
|
|
)
|
|
|
|
articles.append(article)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error parsing Yahoo Finance article: {e}")
|
|
continue
|
|
|
|
logger.info(f"Retrieved {len(articles)} articles from Yahoo Finance")
|
|
return articles
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching Yahoo Finance news for {ticker}: {e}")
|
|
return []
|
|
|
|
async def _get_newsapi_articles(self, ticker: str, days_back: int, max_articles: int) -> List[NewsArticle]:
|
|
"""Get news articles from NewsAPI"""
|
|
try:
|
|
# Rate limiting
|
|
await self._rate_limit_newsapi()
|
|
|
|
logger.info(f"Fetching NewsAPI articles for {ticker}")
|
|
|
|
# Calculate date range
|
|
from_date = (datetime.now() - timedelta(days=days_back)).strftime('%Y-%m-%d')
|
|
|
|
# NewsAPI endpoint
|
|
url = "https://newsapi.org/v2/everything"
|
|
params = {
|
|
"q": f'"{ticker}" OR "{ticker} stock" OR "{ticker} earnings"',
|
|
"from": from_date,
|
|
"sortBy": "publishedAt",
|
|
"pageSize": max_articles,
|
|
"apiKey": self.newsapi_key,
|
|
"language": "en"
|
|
}
|
|
|
|
session = await get_http_session()
|
|
async with session.get(url, params=params) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
|
|
articles = []
|
|
for item in data.get('articles', []):
|
|
try:
|
|
# Parse NewsAPI format - make timezone naive for consistency
|
|
published_str = item['publishedAt'].replace('Z', '+00:00')
|
|
published_at = datetime.fromisoformat(published_str).replace(tzinfo=None)
|
|
|
|
article = NewsArticle(
|
|
title=item.get('title', ''),
|
|
summary=item.get('description', ''),
|
|
content=item.get('content', ''),
|
|
url=item.get('url', ''),
|
|
source="NewsAPI",
|
|
published_at=published_at,
|
|
author=item.get('author', ''),
|
|
image_url=item.get('urlToImage', '')
|
|
)
|
|
|
|
articles.append(article)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error parsing NewsAPI article: {e}")
|
|
continue
|
|
|
|
logger.info(f"Retrieved {len(articles)} articles from NewsAPI")
|
|
return articles
|
|
|
|
else:
|
|
error_data = await response.json()
|
|
logger.error(f"NewsAPI error {response.status}: {error_data}")
|
|
raise NewsAPIError(f"NewsAPI returned {response.status}: {error_data}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching NewsAPI articles for {ticker}: {e}")
|
|
return []
|
|
|
|
async def _get_reddit_posts(self, ticker: str, days_back: int, max_posts: int) -> List[SocialPost]:
|
|
"""Get posts from Reddit related to the ticker"""
|
|
try:
|
|
# Get Reddit access token
|
|
await self._ensure_reddit_token()
|
|
|
|
# Rate limiting
|
|
await self._rate_limit_reddit()
|
|
|
|
logger.info(f"Fetching Reddit posts for {ticker}")
|
|
|
|
# Search multiple relevant subreddits
|
|
subreddits = [
|
|
"stocks", "investing", "SecurityAnalysis", "StockMarket",
|
|
"ValueInvesting", "financialindependence", "wallstreetbets"
|
|
]
|
|
|
|
all_posts = []
|
|
|
|
for subreddit in subreddits:
|
|
try:
|
|
await self._rate_limit_reddit()
|
|
|
|
# Search for ticker in subreddit
|
|
url = f"https://oauth.reddit.com/r/{subreddit}/search"
|
|
params = {
|
|
"q": f'"{ticker}" OR "${ticker}" OR "{ticker} stock"',
|
|
"restrict_sr": "true",
|
|
"sort": "hot",
|
|
"limit": max_posts // len(subreddits) + 1,
|
|
"t": "week" if days_back <= 7 else "month"
|
|
}
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {self._reddit_token}",
|
|
"User-Agent": "StockOracle/1.0.0"
|
|
}
|
|
|
|
session = await get_http_session()
|
|
async with session.get(url, params=params, headers=headers) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
|
|
for item in data.get('data', {}).get('children', []):
|
|
try:
|
|
post_data = item.get('data', {})
|
|
|
|
# Filter out posts that are too old
|
|
created_utc = post_data.get('created_utc', 0)
|
|
post_date = datetime.fromtimestamp(created_utc)
|
|
|
|
if (datetime.now() - post_date).days > days_back:
|
|
continue
|
|
|
|
# Skip removed/deleted posts
|
|
if post_data.get('removed_by_category') or post_data.get('selftext') == '[removed]':
|
|
continue
|
|
|
|
post = SocialPost(
|
|
title=post_data.get('title', ''),
|
|
content=post_data.get('selftext', ''),
|
|
url=f"https://reddit.com{post_data.get('permalink', '')}",
|
|
platform="Reddit",
|
|
author=post_data.get('author', ''),
|
|
published_at=post_date,
|
|
score=post_data.get('score', 0),
|
|
comments_count=post_data.get('num_comments', 0),
|
|
upvotes=post_data.get('ups', 0),
|
|
downvotes=post_data.get('downs', 0),
|
|
subreddit=post_data.get('subreddit', '')
|
|
)
|
|
|
|
all_posts.append(post)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error parsing Reddit post: {e}")
|
|
continue
|
|
|
|
elif response.status == 401:
|
|
logger.error("Reddit API authentication failed")
|
|
# Try to refresh token
|
|
self._reddit_token = None
|
|
await self._ensure_reddit_token()
|
|
else:
|
|
logger.warning(f"Reddit API error for r/{subreddit}: {response.status}")
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error fetching from r/{subreddit}: {e}")
|
|
continue
|
|
|
|
# Remove duplicates based on URL
|
|
seen_urls = set()
|
|
unique_posts = []
|
|
for post in all_posts:
|
|
if post.url not in seen_urls:
|
|
seen_urls.add(post.url)
|
|
unique_posts.append(post)
|
|
|
|
# Sort by score and limit
|
|
unique_posts.sort(key=lambda x: x.score or 0, reverse=True)
|
|
unique_posts = unique_posts[:max_posts]
|
|
|
|
logger.info(f"Retrieved {len(unique_posts)} posts from Reddit")
|
|
return unique_posts
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching Reddit posts for {ticker}: {e}")
|
|
return []
|
|
|
|
async def _ensure_reddit_token(self):
|
|
"""Ensure we have a valid Reddit access token"""
|
|
if self._reddit_token and self._reddit_token_expiry and datetime.now() < self._reddit_token_expiry:
|
|
return
|
|
|
|
logger.info("Obtaining Reddit access token")
|
|
|
|
try:
|
|
# Reddit OAuth2 client credentials flow
|
|
auth_url = "https://www.reddit.com/api/v1/access_token"
|
|
|
|
auth_data = {
|
|
"grant_type": "client_credentials"
|
|
}
|
|
|
|
headers = {
|
|
"User-Agent": "StockOracle/1.0.0"
|
|
}
|
|
|
|
auth = aiohttp.BasicAuth(self.reddit_client_id, self.reddit_client_secret)
|
|
|
|
session = await get_http_session()
|
|
async with session.post(auth_url, data=auth_data, auth=auth, headers=headers) as response:
|
|
if response.status == 200:
|
|
token_data = await response.json()
|
|
|
|
self._reddit_token = token_data.get('access_token')
|
|
expires_in = token_data.get('expires_in', 3600)
|
|
self._reddit_token_expiry = datetime.now() + timedelta(seconds=expires_in - 60)
|
|
|
|
logger.info("Successfully obtained Reddit access token")
|
|
|
|
else:
|
|
error_data = await response.text()
|
|
logger.error(f"Reddit auth error {response.status}: {error_data}")
|
|
raise RedditAPIError(f"Failed to authenticate with Reddit: {response.status}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error obtaining Reddit token: {e}")
|
|
raise
|
|
|
|
async def _rate_limit_newsapi(self):
|
|
"""Apply rate limiting for NewsAPI"""
|
|
now = time.time()
|
|
time_since_last = now - self._last_newsapi_request
|
|
if time_since_last < self._newsapi_rate_limit:
|
|
await asyncio.sleep(self._newsapi_rate_limit - time_since_last)
|
|
self._last_newsapi_request = time.time()
|
|
|
|
async def _rate_limit_reddit(self):
|
|
"""Apply rate limiting for Reddit API"""
|
|
now = time.time()
|
|
time_since_last = now - self._last_reddit_request
|
|
if time_since_last < self._reddit_rate_limit:
|
|
await asyncio.sleep(self._reddit_rate_limit - time_since_last)
|
|
self._last_reddit_request = time.time()
|
|
|
|
|
|
# Global service instance
|
|
news_social_service = NewsSocialService()
|
|
|
|
|
|
# Export for use in other modules
|
|
__all__ = ["news_social_service", "NewsArticle", "SocialPost", "NewsSocialService"] |