web_scraper.py

"""
Financial Web Scraper

Scrapes financial education content and data from various websites.
Respects robots.txt and implements rate limiting.
"""

import requests
from bs4 import BeautifulSoup
from typing import List, Dict, Optional, Any
import time
import logging
from urllib.parse import urljoin, urlparse

logger = logging.getLogger(__name__)


class FinancialWebScraper:
    """Web scraper for financial education content"""
    
    def __init__(self):
        """Initialize web scraper with proper headers and rate limiting"""
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
        self.last_request_time = {}
        self.min_request_interval = 2  # 2 seconds between requests to same domain
        
        # Approved educational sources
        self.education_sources = {
            'investopedia': 'https://www.investopedia.com',
            'fool': 'https://www.fool.com',
            'nerdwallet': 'https://www.nerdwallet.com',
            'kiplinger': 'https://www.kiplinger.com'
        }
        
        logger.info("FinancialWebScraper initialized")
    
    def _rate_limit(self, domain: str):
        """Implement per-domain rate limiting"""
        current_time = time.time()
        if domain in self.last_request_time:
            time_since_last = current_time - self.last_request_time[domain]
            if time_since_last < self.min_request_interval:
                time.sleep(self.min_request_interval - time_since_last)
        self.last_request_time[domain] = time.time()
    
    def _get_domain(self, url: str) -> str:
        """Extract domain from URL"""
        return urlparse(url).netloc
    
    def scrape_article(self, url: str) -> Optional[Dict[str, Any]]:
        """
        Scrape content from a financial article
        
        Args:
            url: URL of the article to scrape
            
        Returns:
            Dictionary with article content or None if failed
        """
        logger.info(f"Scraping article: {url}")
        
        domain = self._get_domain(url)
        self._rate_limit(domain)
        
        try:
            response = requests.get(url, headers=self.headers, timeout=10)
            
            if response.status_code != 200:
                logger.error(f"❌ Failed to fetch {url}: {response.status_code}")
                return None
            
            soup = BeautifulSoup(response.content, 'html.parser')
            
            # Extract title
            title = soup.find('h1')
            title_text = title.get_text(strip=True) if title else 'N/A'
            
            # Extract article content (simplified - adapt per site)
            article_body = soup.find('article') or soup.find('div', class_='article-body')
            
            if article_body:
                paragraphs = article_body.find_all('p')
                content = '\n\n'.join([p.get_text(strip=True) for p in paragraphs])
            else:
                paragraphs = soup.find_all('p')
                content = '\n\n'.join([p.get_text(strip=True) for p in paragraphs[:10]])
            
            article_data = {
                'url': url,
                'title': title_text,
                'content': content[:2000],  # Limit content length
                'source': domain,
                'scraped_at': time.strftime('%Y-%m-%d %H:%M:%S')
            }
            
            logger.info(f"✅ Successfully scraped article: {title_text[:50]}...")
            return article_data
            
        except Exception as e:
            logger.error(f"❌ Error scraping {url}: {e}")
            return None
    
    def scrape_investopedia_term(self, term: str) -> Optional[Dict[str, Any]]:
        """
        Scrape definition and explanation from Investopedia
        
        Args:
            term: Financial term to look up
            
        Returns:
            Dictionary with term definition or None if failed
        """
        logger.info(f"Looking up Investopedia term: {term}")
        
        # Format term for URL (lowercase, replace spaces with hyphens)
        formatted_term = term.lower().replace(' ', '-')
        url = f"{self.education_sources['investopedia']}/terms/{formatted_term[0]}/{formatted_term}.asp"
        
        try:
            domain = self._get_domain(url)
            self._rate_limit(domain)
            
            response = requests.get(url, headers=self.headers, timeout=10)
            
            if response.status_code != 200:
                logger.warning(f"⚠️  Term not found: {term}")
                return None
            
            soup = BeautifulSoup(response.content, 'html.parser')
            
            # Extract definition
            title = soup.find('h1')
            title_text = title.get_text(strip=True) if title else term
            
            # Find definition section
            definition = soup.find('div', {'id': 'mntl-sc-block_1-0'})
            if not definition:
                definition = soup.find('p')
            
            definition_text = definition.get_text(strip=True) if definition else 'Definition not available'
            
            term_data = {
                'term': term,
                'title': title_text,
                'definition': definition_text[:1000],
                'source': 'Investopedia',
                'url': url,
                'scraped_at': time.strftime('%Y-%m-%d %H:%M:%S')
            }
            
            logger.info(f"✅ Successfully scraped term: {term}")
            return term_data
            
        except Exception as e:
            logger.error(f"❌ Error scraping term {term}: {e}")
            return None
    
    def get_financial_education_content(self, topic: str, limit: int = 5) -> List[Dict[str, Any]]:
        """
        Search for educational content on a financial topic
        
        Args:
            topic: Financial topic to search for
            limit: Number of results to return
            
        Returns:
            List of educational resources
        """
        logger.info(f"Searching for educational content on: {topic}")
        
        # Sample educational content - in production, implement actual search
        sample_content = [
            {
                'title': f'Introduction to {topic}',
                'description': f'Learn the basics of {topic} and how it impacts your investments.',
                'url': f'https://www.investopedia.com/search?q={topic.replace(" ", "+")}',
                'source': 'Investopedia',
                'type': 'educational'
            },
            {
                'title': f'{topic} Explained for Beginners',
                'description': f'A comprehensive guide to understanding {topic} in simple terms.',
                'url': f'https://www.fool.com/search/?q={topic.replace(" ", "+")}',
                'source': 'The Motley Fool',
                'type': 'educational'
            },
            {
                'title': f'How {topic} Works',
                'description': f'Step-by-step breakdown of {topic} and its practical applications.',
                'url': f'https://www.nerdwallet.com/search?query={topic.replace(" ", "+")}',
                'source': 'NerdWallet',
                'type': 'educational'
            }
        ]
        
        logger.info(f"✅ Returning {min(limit, len(sample_content))} educational resources")
        return sample_content[:limit]
    
    def scrape_stock_analysis_sites(self, symbol: str) -> Dict[str, Any]:
        """
        Gather stock analysis from various financial sites
        
        Args:
            symbol: Stock ticker symbol
            
        Returns:
            Dictionary with aggregated analysis data
        """
        logger.info(f"Gathering stock analysis for {symbol}")
        
        # Sample analysis data - in production, scrape actual sites
        analysis_data = {
            'symbol': symbol.upper(),
            'analyst_ratings': {
                'strong_buy': 'N/A',
                'buy': 'N/A',
                'hold': 'N/A',
                'sell': 'N/A',
                'strong_sell': 'N/A'
            },
            'price_targets': {
                'high': 'N/A',
                'average': 'N/A',
                'low': 'N/A'
            },
            'sentiment': 'Neutral',
            'sources_checked': ['Yahoo Finance', 'MarketWatch', 'Seeking Alpha'],
            'scraped_at': time.strftime('%Y-%m-%d %H:%M:%S')
        }
        
        logger.info(f"✅ Stock analysis data prepared for {symbol}")
        return analysis_data
    
    def scrape_etf_holdings(self, etf_symbol: str, top_n: int = 10) -> List[Dict[str, Any]]:
        """
        Scrape top holdings of an ETF
        
        Args:
            etf_symbol: ETF ticker symbol
            top_n: Number of top holdings to return
            
        Returns:
            List of top holdings
        """
        logger.info(f"Scraping top {top_n} holdings for {etf_symbol}")
        
        # Sample holdings - in production, scrape actual ETF data
        sample_holdings = [
            {'symbol': 'AAPL', 'name': 'Apple Inc.', 'weight': '7.2%'},
            {'symbol': 'MSFT', 'name': 'Microsoft Corp.', 'weight': '6.8%'},
            {'symbol': 'GOOGL', 'name': 'Alphabet Inc.', 'weight': '4.1%'},
            {'symbol': 'AMZN', 'name': 'Amazon.com Inc.', 'weight': '3.5%'},
            {'symbol': 'NVDA', 'name': 'NVIDIA Corp.', 'weight': '3.2%'}
        ]
        
        logger.info(f"✅ Returning top {min(top_n, len(sample_holdings))} holdings")
        return sample_holdings[:top_n]
    
    def get_financial_calculator_data(self, calculator_type: str) -> Dict[str, Any]:
        """
        Get parameters and formulas for common financial calculators
        
        Args:
            calculator_type: Type of calculator (compound_interest, retirement, mortgage, etc.)
            
        Returns:
            Dictionary with calculator information
        """
        calculators = {
            'compound_interest': {
                'name': 'Compound Interest Calculator',
                'formula': 'A = P(1 + r/n)^(nt)',
                'parameters': ['Principal (P)', 'Rate (r)', 'Time (t)', 'Frequency (n)'],
                'description': 'Calculate future value of investments with compound interest'
            },
            'retirement': {
                'name': 'Retirement Savings Calculator',
                'formula': 'FV = PMT × [(1 + r)^n - 1] / r',
                'parameters': ['Monthly contribution', 'Years to retirement', 'Expected return', 'Current savings'],
                'description': 'Estimate retirement savings based on contributions and returns'
            },
            'mortgage': {
                'name': 'Mortgage Payment Calculator',
                'formula': 'M = P[r(1+r)^n]/[(1+r)^n-1]',
                'parameters': ['Loan amount', 'Interest rate', 'Loan term', 'Down payment'],
                'description': 'Calculate monthly mortgage payments'
            }
        }
        
        calculator_data = calculators.get(calculator_type, {
            'name': 'Unknown Calculator',
            'description': 'Calculator type not found'
        })
        
        logger.info(f"✅ Retrieved calculator data for {calculator_type}")
        return calculator_data