news_data_integration.py

"""News Data Integration - Real-Time Sentiment Pipeline

Connects to real news APIs and RSS feeds for live sentiment signals.
Replaces synthetic news with actual financial headlines.

Supports:
- NewsAPI (https://newsapi.org/) - Free tier available
- RSS feeds (Yahoo Finance, Seeking Alpha, etc.)
- GDELT Project (global news database)
- Reddit/StockTwits social feeds
"""
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
import time
import warnings
warnings.filterwarnings('ignore')

try:
    import feedparser
    FEEDPARSER_AVAILABLE = True
except ImportError:
    FEEDPARSER_AVAILABLE = False


class NewsAPIClient:
    """
    NewsAPI.org client for financial news retrieval.
    
    Free tier: 100 requests/day
    Paid tier: $449/month for 1M requests
    
    Use free tier for prototyping, upgrade for production.
    """
    
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key
        self.base_url = "https://newsapi.org/v2"
        self.last_request_time = 0
        self.min_interval = 1.2  # Free tier: ~80 requests/minute max
    
    def _rate_limit(self):
        """Enforce rate limiting"""
        elapsed = time.time() - self.last_request_time
        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)
        self.last_request_time = time.time()
    
    def fetch_everything(self, 
                         query: str,
                         from_date: Optional[str] = None,
                         to_date: Optional[str] = None,
                         language: str = 'en',
                         sort_by: str = 'publishedAt',
                         page_size: int = 100,
                         page: int = 1) -> List[Dict]:
        """
        Fetch news articles matching query.
        
        Args:
            query: Search query (e.g., "AAPL Apple stock earnings")
            from_date: Start date (YYYY-MM-DD)
            to_date: End date (YYYY-MM-DD)
            language: 'en', 'es', 'fr', etc.
            sort_by: 'relevancy', 'popularity', 'publishedAt'
            page_size: Max 100
            page: Page number
        """
        if self.api_key is None:
            print("WARNING: No API key provided. Using mock data.")
            return self._mock_news(query)
        
        try:
            import requests
        except ImportError:
            print("WARNING: requests library not available. Using mock data.")
            return self._mock_news(query)
        
        self._rate_limit()
        
        params = {
            'q': query,
            'apiKey': self.api_key,
            'language': language,
            'sortBy': sort_by,
            'pageSize': min(page_size, 100),
            'page': page
        }
        
        if from_date:
            params['from'] = from_date
        if to_date:
            params['to'] = to_date
        
        try:
            response = requests.get(
                f"{self.base_url}/everything", 
                params=params,
                timeout=30
            )
            response.raise_for_status()
            data = response.json()
            
            if data.get('status') != 'ok':
                print(f"API Error: {data.get('message', 'Unknown error')}")
                return self._mock_news(query)
            
            articles = data.get('articles', [])
            
            return [{
                'title': a.get('title', ''),
                'description': a.get('description', ''),
                'content': a.get('content', ''),
                'published_at': a.get('publishedAt', ''),
                'source': a.get('source', {}).get('name', 'Unknown'),
                'url': a.get('url', ''),
                'author': a.get('author', '')
            } for a in articles]
            
        except Exception as e:
            print(f"Error fetching news: {e}")
            return self._mock_news(query)
    
    def fetch_for_ticker(self, 
                          ticker: str,
                          company_name: str,
                          from_date: Optional[str] = None,
                          to_date: Optional[str] = None,
                          page_size: int = 100) -> pd.DataFrame:
        """
        Fetch news for a specific ticker and return formatted DataFrame.
        """
        query = f"{ticker} {company_name} stock"
        articles = self.fetch_everything(
            query=query,
            from_date=from_date,
            to_date=to_date,
            page_size=page_size
        )
        
        df = pd.DataFrame(articles)
        df['ticker'] = ticker
        df['query'] = query
        
        # Combine title and content for analysis
        df['text'] = df['title'].fillna('') + '. ' + df['description'].fillna('') + ' ' + df['content'].fillna('')
        df['text'] = df['text'].str.strip()
        
        # Parse dates
        df['date'] = pd.to_datetime(df['published_at'], errors='coerce')
        
        return df
    
    def fetch_multiple_tickers(self,
                              ticker_map: Dict[str, str],
                              from_date: Optional[str] = None,
                              to_date: Optional[str] = None) -> pd.DataFrame:
        """
        Fetch news for multiple tickers.
        
        Args:
            ticker_map: {ticker: company_name}
        """
        all_news = []
        
        for ticker, company in ticker_map.items():
            print(f"Fetching news for {ticker} ({company})...")
            try:
                df = self.fetch_for_ticker(
                    ticker, company, from_date, to_date
                )
                all_news.append(df)
            except Exception as e:
                print(f"  Error for {ticker}: {e}")
        
        if all_news:
            return pd.concat(all_news, ignore_index=True)
        return pd.DataFrame()
    
    def _mock_news(self, query: str) -> List[Dict]:
        """Generate mock news for testing without API key"""
        import random
        
        templates = [
            {"title": "{company} reports strong quarterly earnings, beating expectations",
             "sentiment": "positive"},
            {"title": "{company} faces regulatory scrutiny over data practices",
             "sentiment": "negative"},
            {"title": "Analysts upgrade {company} to buy rating",
             "sentiment": "positive"},
            {"title": "{company} announces major product launch",
             "sentiment": "positive"},
            {"title": "{company} stock falls amid market volatility",
             "sentiment": "negative"},
            {"title": "Market awaits {company} earnings report next week",
             "sentiment": "neutral"},
        ]
        
        company = query.split()[0] if query else "Company"
        
        articles = []
        for i, template in enumerate(random.sample(templates, min(3, len(templates)))):
            articles.append({
                'title': template['title'].format(company=company),
                'description': f"Analysis of {company} stock performance.",
                'content': f"Detailed article about {company} and market conditions.",
                'published_at': (datetime.now() - timedelta(hours=i*6)).isoformat(),
                'source': f'MockSource{i}',
                'url': f'https://example.com/article{i}',
                'author': 'MockAuthor'
            })
        
        return articles


class RSSFeedClient:
    """
    RSS Feed client for real-time financial news.
    
    No API key needed! Just RSS feeds from financial websites.
    """
    
    FINANCIAL_FEEDS = {
        'yahoo_finance': 'https://finance.yahoo.com/news/rssindex',
        'marketwatch': 'https://www.marketwatch.com/rss/topstories',
        'seeking_alpha': 'https://seekingalpha.com/market_currents.xml',
        'investing_com': 'https://www.investing.com/rss/news.rss',
        'barrons': 'https://www.barrons.com/articles/rss',
        'wall_street_journal': 'https://feeds.a.dj.com/rss/WSJcomUSBusiness.xml',
        'reuters_business': 'https://www.reutersagency.com/feed/?taxonomy=markets&post_type=reuters-best',
        'benzinga': 'https://www.benzinga.com/feed',
        'the_street': 'https://www.thestreet.com/.rss/full/',
    }
    
    def __init__(self):
        self.feeds = {}
    
    def fetch_feed(self, feed_url: str, max_entries: int = 50) -> List[Dict]:
        """Fetch and parse an RSS feed"""
        if not FEEDPARSER_AVAILABLE:
            print("WARNING: feedparser not available. Install with: pip install feedparser")
            return []
        
        try:
            feed = feedparser.parse(feed_url)
            
            articles = []
            for entry in feed.entries[:max_entries]:
                articles.append({
                    'title': entry.get('title', ''),
                    'description': entry.get('summary', entry.get('description', '')),
                    'content': entry.get('summary', ''),
                    'published_at': entry.get('published', entry.get('updated', '')),
                    'source': feed.feed.get('title', 'Unknown'),
                    'url': entry.get('link', ''),
                    'author': entry.get('author', '')
                })
            
            return articles
            
        except Exception as e:
            print(f"Error fetching RSS feed {feed_url}: {e}")
            return []
    
    def fetch_all_feeds(self, max_entries_per_feed: int = 20) -> pd.DataFrame:
        """Fetch all configured financial feeds"""
        all_articles = []
        
        for name, url in self.FINANCIAL_FEEDS.items():
            print(f"Fetching {name}...")
            articles = self.fetch_feed(url, max_entries_per_feed)
            for a in articles:
                a['feed_source'] = name
            all_articles.extend(articles)
        
        if not all_articles:
            return pd.DataFrame()
        
        df = pd.DataFrame(all_articles)
        df['text'] = df['title'].fillna('') + '. ' + df['description'].fillna('')
        df['date'] = pd.to_datetime(df['published_at'], errors='coerce')
        
        return df
    
    def add_custom_feed(self, name: str, url: str):
        """Add a custom RSS feed"""
        self.FINANCIAL_FEEDS[name] = url


class GDELTClient:
    """
    GDELT Project (Global Database of Events, Language, and Tone) client.
    
    Free, massive global news database.
    https://www.gdeltproject.org/
    
    GDELT provides:
    - Every news article worldwide (updated every 15 minutes)
    - Sentiment scoring (tone)
    - Event coding
    - Geographic tagging
    """
    
    GDELT_URL = "http://data.gdeltproject.org/gdeltv2/lastupdate.txt"
    
    def __init__(self):
        pass
    
    def fetch_latest_updates(self) -> pd.DataFrame:
        """Fetch latest GDELT update URLs"""
        try:
            import requests
            response = requests.get(self.GDELT_URL, timeout=30)
            response.raise_for_status()
            
            lines = response.text.strip().split('\n')
            
            updates = []
            for line in lines:
                parts = line.split()
                if len(parts) >= 3:
                    updates.append({
                        'timestamp': parts[0],
                        'url': parts[2],
                        'type': 'events' if 'export' in parts[2] else 'mentions' if 'mentions' in parts[2] else 'gkg'
                    })
            
            return pd.DataFrame(updates)
            
        except Exception as e:
            print(f"Error fetching GDELT updates: {e}")
            return pd.DataFrame()
    
    def fetch_gdelt_csv(self, url: str) -> pd.DataFrame:
        """Fetch and parse a GDELT CSV file"""
        try:
            import requests
            import zipfile
            import io
            
            response = requests.get(url, timeout=60)
            response.raise_for_status()
            
            # GDELT files are ZIP archives
            with zipfile.ZipFile(io.BytesIO(response.content)) as z:
                csv_name = z.namelist()[0]
                with z.open(csv_name) as f:
                    df = pd.read_csv(f, sep='\t', header=None, low_memory=False)
            
            # Add column names based on type
            if 'export' in url:
                # CAMEO event data
                columns = ['GlobalEventID', 'Day', 'MonthYear', 'Year', 'FractionDate',
                          'Actor1Code', 'Actor1Name', 'Actor1CountryCode', 'Actor1KnownGroupCode',
                          'Actor1EthnicCode', 'Actor1Religion1Code', 'Actor1Religion2Code',
                          'Actor1Type1Code', 'Actor1Type2Code', 'Actor1Type3Code',
                          'Actor2Code', 'Actor2Name', 'Actor2CountryCode', 'Actor2KnownGroupCode',
                          'Actor2EthnicCode', 'Actor2Religion1Code', 'Actor2Religion2Code',
                          'Actor2Type1Code', 'Actor2Type2Code', 'Actor2Type3Code',
                          'IsRootEvent', 'EventCode', 'EventBaseCode', 'EventRootCode',
                          'QuadClass', 'GoldsteinScale', 'NumMentions', 'NumSources',
                          'NumArticles', 'AvgTone', 'Actor1Geo_Type', 'Actor1Geo_FullName',
                          'Actor1Geo_CountryCode', 'Actor1Geo_ADM1Code', 'Actor1Geo_Lat',
                          'Actor1Geo_Long', 'Actor1Geo_FeatureID', 'Actor2Geo_Type',
                          'Actor2Geo_FullName', 'Actor2Geo_CountryCode', 'Actor2Geo_ADM1Code',
                          'Actor2Geo_Lat', 'Actor2Geo_Long', 'Actor2Geo_FeatureID',
                          'ActionGeo_Type', 'ActionGeo_FullName', 'ActionGeo_CountryCode',
                          'ActionGeo_ADM1Code', 'ActionGeo_Lat', 'ActionGeo_Long',
                          'ActionGeo_FeatureID', 'DATEADDED', 'SOURCEURL']
                df.columns = columns[:len(df.columns)]
            
            return df
            
        except Exception as e:
            print(f"Error fetching GDELT data: {e}")
            return pd.DataFrame()


class SocialMediaScraper:
    """
    Social media sentiment scraper (Reddit, StockTwits, Twitter/X).
    
    Note: Twitter API now requires paid access ($100/month basic tier).
    Reddit API has rate limits but free tier available.
    StockTwits has free API for basic usage.
    """
    
    REDDIT_SUBREDDITS = [
        'wallstreetbets', 'stocks', 'investing', 'StockMarket',
        'options', 'pennystocks', 'SecurityAnalysis', 'algotrading'
    ]
    
    def __init__(self):
        pass
    
    def fetch_reddit_posts(self,
                           subreddit: str,
                           limit: int = 100,
                           time_filter: str = 'day') -> pd.DataFrame:
        """
        Fetch Reddit posts from a subreddit.
        
        Requires: pip install praw
        You need Reddit API credentials (free at reddit.com/prefs/apps)
        """
        try:
            import praw
        except ImportError:
            print("WARNING: praw not available. Install with: pip install praw")
            return pd.DataFrame()
        
        # Note: User must provide their own credentials
        # This is a placeholder showing the pattern
        print("REDDIT INTEGRATION PATTERN:")
        print("  1. Create app at https://www.reddit.com/prefs/apps")
        print("  2. Get client_id and client_secret")
        print("  3. Initialize: praw.Reddit(client_id='...', client_secret='...', user_agent='...')")
        print("  4. Fetch: reddit.subreddit('wallstreetbets').hot(limit=100)")
        
        return pd.DataFrame()
    
    def fetch_stocktwits_feed(self,
                            ticker: str,
                            limit: int = 30) -> pd.DataFrame:
        """
        Fetch StockTwits messages for a ticker.
        
        StockTwits API: https://api.stocktwits.com/developers/docs
        Free tier available for basic usage.
        """
        try:
            import requests
        except ImportError:
            print("WARNING: requests not available")
            return pd.DataFrame()
        
        url = f"https://api.stocktwits.com/api/2/streams/symbol/{ticker}.json"
        
        try:
            response = requests.get(url, timeout=30)
            response.raise_for_status()
            data = response.json()
            
            messages = data.get('messages', [])
            
            return pd.DataFrame([{
                'text': m.get('body', ''),
                'created_at': m.get('created_at', ''),
                'username': m.get('user', {}).get('username', ''),
                'sentiment': m.get('entities', {}).get('sentiment', {}).get('basic', 'neutral'),
                'likes': m.get('likes', {}).get('total', 0),
                'ticker': ticker
            } for m in messages])
            
        except Exception as e:
            print(f"Error fetching StockTwits: {e}")
            return pd.DataFrame()


class NewsPipeline:
    """
    Complete news pipeline: fetch -> preprocess -> sentiment -> aggregate.
    
    Connects NewsAPI + RSS feeds + Social media into one unified feed.
    """
    
    def __init__(self,
                 news_api_key: Optional[str] = None,
                 use_rss: bool = True,
                 use_gdelt: bool = False,
                 use_social: bool = False):
        self.news_api = NewsAPIClient(news_api_key)
        self.rss_client = RSSFeedClient()
        self.gdelt_client = GDELTClient()
        self.social_scraper = SocialMediaScraper()
        
        self.use_rss = use_rss
        self.use_gdelt = use_gdelt
        self.use_social = use_social
    
    def fetch_all(self,
                  tickers: List[str],
                  company_names: Optional[Dict[str, str]] = None,
                  from_date: Optional[str] = None,
                  to_date: Optional[str] = None) -> pd.DataFrame:
        """
        Fetch news from ALL sources for given tickers.
        
        Returns unified DataFrame with all articles.
        """
        all_news = []
        
        # NewsAPI
        if company_names:
            ticker_map = {t: company_names.get(t, t) for t in tickers}
        else:
            ticker_map = {t: t for t in tickers}
        
        print("[NewsAPI] Fetching financial news...")
        try:
            news_api_df = self.news_api.fetch_multiple_tickers(
                ticker_map, from_date, to_date
            )
            if not news_api_df.empty:
                news_api_df['source_type'] = 'newsapi'
                all_news.append(news_api_df)
        except Exception as e:
            print(f"  NewsAPI error: {e}")
        
        # RSS Feeds
        if self.use_rss:
            print("[RSS] Fetching financial feeds...")
            try:
                rss_df = self.rss_client.fetch_all_feeds(max_entries_per_feed=10)
                if not rss_df.empty:
                    rss_df['source_type'] = 'rss'
                    # Tag with tickers using simple keyword matching
                    rss_df['ticker'] = rss_df['text'].apply(
                        lambda x: self._extract_tickers(str(x), tickers)
                    )
                    rss_df = rss_df[rss_df['ticker'].notna()]
                    if not rss_df.empty:
                        all_news.append(rss_df)
            except Exception as e:
                print(f"  RSS error: {e}")
        
        # Combine
        if all_news:
            combined = pd.concat(all_news, ignore_index=True)
            combined['text'] = combined['text'].fillna('')
            combined['date'] = pd.to_datetime(combined['date'], errors='coerce')
            return combined.sort_values('date', ascending=False)
        
        return pd.DataFrame()
    
    def _extract_tickers(self, text: str, tickers: List[str]) -> Optional[str]:
        """Simple keyword matching to tag articles with tickers"""
        text_upper = text.upper()
        for ticker in tickers:
            if f' {ticker} ' in text_upper or f'${ticker}' in text_upper:
                return ticker
        return None
    
    def aggregate_daily_sentiment(self,
                                   news_df: pd.DataFrame,
                                   sentiment_fn: Optional[Callable] = None) -> pd.DataFrame:
        """
        Aggregate news into daily sentiment scores per ticker.
        
        Requires sentiment_fn that takes text -> dict with 'sentiment_score'.
        If not provided, returns raw counts only.
        """
        if news_df.empty:
            return pd.DataFrame()
        
        # Ensure date is datetime
        news_df['date'] = pd.to_datetime(news_df['date'], errors='coerce')
        news_df['date'] = news_df['date'].dt.date
        
        if sentiment_fn is not None:
            print("Computing sentiment scores...")
            sentiments = []
            for text in news_df['text']:
                try:
                    result = sentiment_fn(str(text))
                    sentiments.append(result.get('sentiment_score', 0))
                except:
                    sentiments.append(0)
            news_df['sentiment_score'] = sentiments
        else:
            news_df['sentiment_score'] = 0
        
        # Aggregate by date and ticker
        daily = news_df.groupby(['date', 'ticker']).agg({
            'sentiment_score': ['mean', 'std', 'count'],
            'text': 'first'
        }).reset_index()
        
        # Flatten multi-index columns
        daily.columns = ['date', 'ticker', 'sentiment_mean', 'sentiment_std',
                        'article_count', 'sample_text']
        
        # Confidence weighting: more articles = more confident
        daily['confidence'] = np.minimum(daily['article_count'] / 5, 1.0)
        daily['sentiment_alpha'] = daily['sentiment_mean'] * daily['confidence']
        
        return daily


if __name__ == '__main__':
    # Test news pipeline
    pipeline = NewsPipeline()
    
    # Fetch mock news (no API key)
    news = pipeline.news_api.fetch_for_ticker('AAPL', 'Apple', page_size=5)
    print(f"Fetched {len(news)} articles for AAPL")
    print(news[['title', 'source', 'date']].head())