Yahoo Finance Scraping with Ollama: Free Stock Data Analysis Tutorial

pip install requests beautifulsoup4 pandas yfinance ollama-python selenium webdriver-manager

ollama pull llama2:7b
# Alternative: ollama pull codellama:7b for code-focused analysis

# src/scraper.py
import requests
from bs4 import BeautifulSoup
import pandas as pd
import yfinance as yf
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
import time
import json

class YahooFinanceScraper:
    def __init__(self, headless=True):
        """Initialize the scraper with optional headless mode"""
        self.base_url = "https://finance.yahoo.com"
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })
        
        # Setup Selenium for dynamic content
        if headless:
            chrome_options = webdriver.ChromeOptions()
            chrome_options.add_argument("--headless")
            chrome_options.add_argument("--no-sandbox")
            chrome_options.add_argument("--disable-dev-shm-usage")
            
            service = Service(ChromeDriverManager().install())
            self.driver = webdriver.Chrome(service=service, options=chrome_options)
    
    def get_stock_data(self, symbol, period="1y"):
        """Fetch historical stock data using yfinance library"""
        try:
            stock = yf.Ticker(symbol)
            data = stock.history(period=period)
            return data
        except Exception as e:
            print(f"Error fetching data for {symbol}: {e}")
            return None
    
    def scrape_financial_metrics(self, symbol):
        """Scrape key financial metrics from Yahoo Finance"""
        url = f"{self.base_url}/quote/{symbol}/key-statistics"
        
        try:
            response = self.session.get(url)
            soup = BeautifulSoup(response.content, 'html.parser')
            
            metrics = {}
            
            # Find metric tables
            tables = soup.find_all('table')
            for table in tables:
                rows = table.find_all('tr')
                for row in rows:
                    cells = row.find_all('td')
                    if len(cells) >= 2:
                        metric_name = cells[0].get_text(strip=True)
                        metric_value = cells[1].get_text(strip=True)
                        metrics[metric_name] = metric_value
            
            return metrics
        except Exception as e:
            print(f"Error scraping metrics for {symbol}: {e}")
            return {}
    
    def get_analyst_recommendations(self, symbol):
        """Extract analyst recommendations and price targets"""
        url = f"{self.base_url}/quote/{symbol}/analysis"
        
        self.driver.get(url)
        time.sleep(3)  # Wait for dynamic content to load
        
        try:
            recommendations = {}
            
            # Find recommendation summary
            rec_elements = self.driver.find_elements("css selector", "[data-test='rec-rating-txt']")
            if rec_elements:
                recommendations['current_rating'] = rec_elements[0].text
            
            # Find price targets
            price_elements = self.driver.find_elements("css selector", "[data-test='target-price-val']")
            if price_elements:
                recommendations['price_target'] = price_elements[0].text
            
            return recommendations
        except Exception as e:
            print(f"Error getting recommendations for {symbol}: {e}")
            return {}
    
    def batch_scrape_stocks(self, symbols, save_path="data/raw/"):
        """Scrape multiple stocks and save to files"""
        all_data = {}
        
        for symbol in symbols:
            print(f"Scraping {symbol}...")
            
            stock_info = {
                'historical_data': self.get_stock_data(symbol),
                'financial_metrics': self.scrape_financial_metrics(symbol),
                'analyst_recs': self.get_analyst_recommendations(symbol)
            }
            
            all_data[symbol] = stock_info
            
            # Save individual stock data
            stock_info['historical_data'].to_csv(f"{save_path}{symbol}_history.csv")
            
            # Rate limiting
            time.sleep(2)
        
        return all_data

# src/utils.py
import time
import random
from functools import wraps

def rate_limit(min_delay=1, max_delay=3):
    """Decorator to add random delays between requests"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            delay = random.uniform(min_delay, max_delay)
            time.sleep(delay)
            return func(*args, **kwargs)
        return wrapper
    return decorator

def retry_on_failure(max_retries=3, delay=5):
    """Decorator to retry failed requests"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise e
                    print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay} seconds...")
                    time.sleep(delay)
        return wrapper
    return decorator

# src/analyzer.py
import ollama
import pandas as pd
import json
from typing import Dict, List

class OllamaStockAnalyzer:
    def __init__(self, model_name="llama2:7b"):
        """Initialize connection to Ollama"""
        self.model = model_name
        self.client = ollama.Client()
    
    def analyze_stock_data(self, symbol: str, stock_data: Dict) -> Dict:
        """Comprehensive stock analysis using Ollama"""
        
        # Prepare data summary for AI analysis
        historical_data = stock_data.get('historical_data')
        financial_metrics = stock_data.get('financial_metrics', {})
        analyst_recs = stock_data.get('analyst_recs', {})
        
        # Calculate key indicators
        current_price = historical_data['Close'].iloc[-1] if not historical_data.empty else 0
        price_change = self._calculate_price_change(historical_data)
        volatility = self._calculate_volatility(historical_data)
        
        # Create analysis prompt
        prompt = self._create_analysis_prompt(
            symbol, current_price, price_change, volatility, 
            financial_metrics, analyst_recs
        )
        
        # Get AI analysis
        response = self.client.chat(
            model=self.model,
            messages=[{
                'role': 'user',
                'content': prompt
            }]
        )
        
        return {
            'symbol': symbol,
            'ai_analysis': response['message']['content'],
            'key_metrics': {
                'current_price': current_price,
                'price_change_1d': price_change,
                'volatility_30d': volatility
            }
        }
    
    def _create_analysis_prompt(self, symbol, price, change, volatility, metrics, recs):
        """Create a detailed prompt for stock analysis"""
        prompt = f"""
        Analyze the stock {symbol} based on the following data:
        
        Current Price: ${price:.2f}
        1-Day Change: {change:.2f}%
        30-Day Volatility: {volatility:.2f}%
        
        Financial Metrics:
        {json.dumps(metrics, indent=2)}
        
        Analyst Recommendations:
        {json.dumps(recs, indent=2)}
        
        Provide a detailed analysis covering:
        1. Technical analysis of price trends
        2. Fundamental valuation assessment
        3. Risk factors and opportunities
        4. Investment recommendation (Buy/Hold/Sell)
        5. Price target and timeline
        
        Format your response as structured analysis with clear sections.
        """
        return prompt
    
    def _calculate_price_change(self, data: pd.DataFrame) -> float:
        """Calculate 1-day price change percentage"""
        if len(data) < 2:
            return 0.0
        return ((data['Close'].iloc[-1] - data['Close'].iloc[-2]) / data['Close'].iloc[-2]) * 100
    
    def _calculate_volatility(self, data: pd.DataFrame, window=30) -> float:
        """Calculate rolling volatility"""
        if len(data) < window:
            window = len(data)
        returns = data['Close'].pct_change().dropna()
        return returns.rolling(window=window).std().iloc[-1] * 100
    
    def generate_portfolio_analysis(self, portfolio_data: Dict) -> str:
        """Analyze entire portfolio performance"""
        portfolio_prompt = f"""
        Analyze this stock portfolio:
        
        {json.dumps(portfolio_data, indent=2, default=str)}
        
        Provide insights on:
        1. Portfolio diversification
        2. Risk assessment
        3. Performance analysis
        4. Rebalancing recommendations
        5. Sector allocation suggestions
        """
        
        response = self.client.chat(
            model=self.model,
            messages=[{'role': 'user', 'content': portfolio_prompt}]
        )
        
        return response['message']['content']

def technical_analysis(self, data: pd.DataFrame) -> Dict:
    """Generate technical indicators and patterns"""
    analysis = {}
    
    # Moving averages
    data['MA_20'] = data['Close'].rolling(window=20).mean()
    data['MA_50'] = data['Close'].rolling(window=50).mean()
    
    # RSI calculation
    delta = data['Close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
    rs = gain / loss
    data['RSI'] = 100 - (100 / (1 + rs))
    
    # MACD
    exp1 = data['Close'].ewm(span=12).mean()
    exp2 = data['Close'].ewm(span=26).mean()
    data['MACD'] = exp1 - exp2
    data['Signal'] = data['MACD'].ewm(span=9).mean()
    
    # Support and resistance levels
    recent_high = data['High'].rolling(window=20).max().iloc[-1]
    recent_low = data['Low'].rolling(window=20).min().iloc[-1]
    
    analysis['indicators'] = {
        'rsi_current': data['RSI'].iloc[-1],
        'macd_signal': 'bullish' if data['MACD'].iloc[-1] > data['Signal'].iloc[-1] else 'bearish',
        'ma_trend': 'uptrend' if data['MA_20'].iloc[-1] > data['MA_50'].iloc[-1] else 'downtrend',
        'resistance': recent_high,
        'support': recent_low
    }
    
    return analysis

# main.py
from src.scraper import YahooFinanceScraper
from src.analyzer import OllamaStockAnalyzer
import json

def main():
    # Initialize components
    scraper = YahooFinanceScraper()
    analyzer = OllamaStockAnalyzer()
    
    # Define stocks to analyze
    symbols = ['AAPL', 'GOOGL', 'MSFT', 'TSLA', 'NVDA']
    
    print("Starting Yahoo Finance scraping with Ollama analysis...")
    
    # Scrape stock data
    stock_data = scraper.batch_scrape_stocks(symbols)
    
    # Analyze each stock with AI
    analysis_results = {}
    for symbol, data in stock_data.items():
        print(f"Analyzing {symbol} with Ollama...")
        analysis = analyzer.analyze_stock_data(symbol, data)
        analysis_results[symbol] = analysis
        
        # Save individual analysis
        with open(f"data/processed/{symbol}_analysis.json", 'w') as f:
            json.dump(analysis, f, indent=2, default=str)
    
    # Generate portfolio overview
    portfolio_analysis = analyzer.generate_portfolio_analysis(analysis_results)
    
    # Save complete results
    final_report = {
        'portfolio_analysis': portfolio_analysis,
        'individual_stocks': analysis_results,
        'timestamp': pd.Timestamp.now().isoformat()
    }
    
    with open("data/processed/complete_analysis.json", 'w') as f:
        json.dump(final_report, f, indent=2, default=str)
    
    print("Analysis complete! Check data/processed/ for results.")
    return final_report

if __name__ == "__main__":
    results = main()

{
  "symbol": "AAPL",
  "ai_analysis": "Apple (AAPL) shows strong technical momentum with price above key moving averages. Current valuation appears reasonable given recent earnings growth...",
  "key_metrics": {
    "current_price": 178.45,
    "price_change_1d": 2.34,
    "volatility_30d": 23.8
  }
}

def scrape_real_time_data(self, symbol):
    """Get real-time quotes that load via JavaScript"""
    url = f"{self.base_url}/quote/{symbol}"
    
    self.driver.get(url)
    
    # Wait for price element to load
    price_element = WebDriverWait(self.driver, 10).until(
        EC.presence_of_element_located((By.CSS_SELECTOR, "[data-test='qsp-price']"))
    )
    
    real_time_data = {
        'price': price_element.text,
        'change': self.driver.find_element(By.CSS_SELECTOR, "[data-test='qsp-price-change']").text,
        'volume': self.driver.find_element(By.CSS_SELECTOR, "[data-test='TD_VOLUME-value']").text
    }
    
    return real_time_data

import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor

class AsyncYahooScraper:
    def __init__(self, max_concurrent=5):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def scrape_multiple_async(self, symbols):
        """Scrape multiple stocks concurrently"""
        async with aiohttp.ClientSession() as session:
            tasks = [self.scrape_single_async(session, symbol) for symbol in symbols]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return dict(zip(symbols, results))
    
    async def scrape_single_async(self, session, symbol):
        """Asynchronous single stock scraping"""
        async with self.semaphore:
            url = f"https://finance.yahoo.com/quote/{symbol}"
            async with session.get(url) as response:
                content = await response.text()
                # Parse content here
                await asyncio.sleep(1)  # Rate limiting
                return content

def validate_stock_data(self, data):
    """Validate scraped stock data quality"""
    validation_results = {
        'valid': True,
        'issues': []
    }
    
    # Check for missing price data
    if data['Close'].isnull().sum() > len(data) * 0.1:
        validation_results['issues'].append("High percentage of missing price data")
        validation_results['valid'] = False
    
    # Check for unrealistic price movements
    daily_changes = data['Close'].pct_change().abs()
    if daily_changes.max() > 0.5:  # 50% daily change threshold
        validation_results['issues'].append("Unrealistic price movements detected")
    
    # Validate volume data
    if (data['Volume'] == 0).sum() > 0:
        validation_results['issues'].append("Zero volume days found")
    
    return validation_results

import gc
from contextlib import contextmanager

@contextmanager
def memory_efficient_processing():
    """Context manager for memory-efficient batch processing"""
    try:
        yield
    finally:
        gc.collect()

def process_large_dataset(self, symbols, batch_size=10):
    """Process large stock lists in batches"""
    results = {}
    
    for i in range(0, len(symbols), batch_size):
        batch = symbols[i:i+batch_size]
        
        with memory_efficient_processing():
            batch_results = self.batch_scrape_stocks(batch)
            results.update(batch_results)
            
            # Save intermediate results
            self.save_batch_results(batch_results, batch_number=i//batch_size)
    
    return results

# scheduler.py
import schedule
import time
from datetime import datetime, timedelta

class AutomatedAnalyzer:
    def __init__(self):
        self.scraper = YahooFinanceScraper()
        self.analyzer = OllamaStockAnalyzer()
        self.watchlist = ['AAPL', 'GOOGL', 'MSFT', 'TSLA', 'NVDA']
    
    def daily_market_analysis(self):
        """Run daily analysis during market hours"""
        if self.is_market_open():
            print(f"Running daily analysis at {datetime.now()}")
            results = self.run_full_analysis()
            self.send_alerts(results)
    
    def is_market_open(self):
        """Check if US market is currently open"""
        now = datetime.now()
        market_open = now.replace(hour=9, minute=30, second=0, microsecond=0)
        market_close = now.replace(hour=16, minute=0, second=0, microsecond=0)
        
        return (market_open <= now <= market_close and 
                now.weekday() < 5)  # Monday = 0, Friday = 4
    
    def run_scheduler(self):
        """Start the automated scheduler"""
        schedule.every().day.at("09:35").do(self.daily_market_analysis)
        schedule.every().day.at("15:55").do(self.daily_market_analysis)
        
        while True:
            schedule.run_pending()
            time.sleep(60)

# Run: python scheduler.py

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    wget \
    curl \
    unzip \
    && rm -rf /var/lib/apt/lists/*

# Install Chrome for Selenium
RUN wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add - \
    && echo "deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main" >> /etc/apt/sources.list.d/google.list \
    && apt-get update \
    && apt-get install -y google-chrome-stable

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

# Install Ollama
RUN curl -fsSL https://ollama.ai/install.sh | sh

CMD ["python", "main.py"]

import pickle
from functools import lru_cache
import hashlib

class CachedScraper(YahooFinanceScraper):
    def __init__(self, cache_dir="cache/"):
        super().__init__()
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)
    
    def get_cached_data(self, symbol, max_age_hours=1):
        """Retrieve cached data if still fresh"""
        cache_file = f"{self.cache_dir}{symbol}_cache.pkl"
        
        if os.path.exists(cache_file):
            with open(cache_file, 'rb') as f:
                cached_data = pickle.load(f)
            
            cache_age = datetime.now() - cached_data['timestamp']
            if cache_age < timedelta(hours=max_age_hours):
                return cached_data['data']
        
        return None
    
    def cache_data(self, symbol, data):
        """Save data to cache with timestamp"""
        cache_file = f"{self.cache_dir}{symbol}_cache.pkl"
        cache_entry = {
            'data': data,
            'timestamp': datetime.now()
        }
        
        with open(cache_file, 'wb') as f:
            pickle.dump(cache_entry, f)

import logging
from collections import defaultdict
import time

class PerformanceMonitor:
    def __init__(self):
        self.metrics = defaultdict(list)
        self.logger = logging.getLogger(__name__)
    
    def time_function(self, func_name):
        """Decorator to time function execution"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                start_time = time.time()
                try:
                    result = func(*args, **kwargs)
                    success = True
                except Exception as e:
                    self.logger.error(f"Function {func_name} failed: {e}")
                    success = False
                    raise
                finally:
                    execution_time = time.time() - start_time
                    self.metrics[func_name].append({
                        'execution_time': execution_time,
                        'success': success,
                        'timestamp': datetime.now()
                    })
                return result
            return wrapper
        return decorator
    
    def get_performance_report(self):
        """Generate performance summary"""
        report = {}
        for func_name, metrics in self.metrics.items():
            avg_time = sum(m['execution_time'] for m in metrics) / len(metrics)
            success_rate = sum(m['success'] for m in metrics) / len(metrics)
            
            report[func_name] = {
                'average_time': avg_time,
                'success_rate': success_rate,
                'total_calls': len(metrics)
            }
        
        return report

class GDPRCompliantScraper:
    def __init__(self):
        self.data_retention_days = 30  # Automatic data deletion
        self.user_consent_required = True
    
    def cleanup_old_data(self):
        """Automatically delete data older than retention period"""
        cutoff_date = datetime.now() - timedelta(days=self.data_retention_days)
        
        for file in os.listdir("data/"):
            file_path = os.path.join("data/", file)
            if os.path.getctime(file_path) < cutoff_date.timestamp():
                os.remove(file_path)
                self.logger.info(f"Deleted old data file: {file}")