Build a Powerful Yahoo Scraper with ScrapeGraphAI: The Ultimate Guide

pip install scrapegraph-py

from scrapegraph_py import Client
from scrapegraph_py.logger import sgai_logger
 
# Enable detailed logging for debugging
sgai_logger.set_logging(level="INFO")
 
# Initialize the client with your API key
sgai_client = Client(api_key="YOUR_API_KEY_HERE")
 
# Define your Yahoo target URLs
yahoo_urls = [
    "https://finance.yahoo.com/quote/AAPL",
    "https://news.yahoo.com/tech",
    "https://yahoo.com/search?q=your-query"
]
 
# SmartScraper request for Yahoo data
response = sgai_client.smartscraper(
    website_url="https://finance.yahoo.com/quote/AAPL",
    user_prompt="Extract stock price, company name, market cap, and recent news headlines"
)
 
print(f"Yahoo Data Results: {response.result}")

import os
from scrapegraph_py import Client
from scrapegraph_py.logger import sgai_logger
import json
from datetime import datetime
 
# Set up logging
sgai_logger.set_logging(level="INFO")
 
# Initialize the client (use environment variable for security)
sgai_client = Client(api_key=os.getenv("SCRAPEGRAPHAI_API_KEY"))
 
def extract_yahoo_finance_data(stock_symbol):
    """
    Extract comprehensive stock data from Yahoo Finance
    
    Args:
        stock_symbol (str): Stock symbol (e.g., AAPL, GOOGL, MSFT)
    
    Returns:
        dict: Structured stock data
    """
    
    yahoo_url = f"https://finance.yahoo.com/quote/{stock_symbol}"
    
    financial_prompt = f"""
    Extract the following comprehensive data from this Yahoo Finance page for {stock_symbol}:
    
    BASIC INFO:
    - Company name
    - Stock symbol
    - Current stock price
    - Currency
    
    PRICE MOVEMENT:
    - Daily change (amount and percentage)
    - Previous close
    - Open price
    - Day's range (high and low)
    - 52-week range (high and low)
    
    VOLUME DATA:
    - Trading volume
    - Average volume
    - Volume ratio
    
    MARKET DATA:
    - Market capitalization
    - Enterprise value
    - Shares outstanding
    - Float shares
    
    VALUATION METRICS:
    - Price-to-earnings (P/E) ratio
    - Price-to-book (P/B) ratio
    - Price-to-sales (P/S) ratio
    - Enterprise value to EBITDA
    - PEG ratio
    
    FINANCIAL RATIOS:
    - Return on equity (ROE)
    - Return on assets (ROA)
    - Debt-to-equity ratio
    - Current ratio
    - Quick ratio
    
    DIVIDEND INFO:
    - Dividend yield
    - Dividend per share
    - Ex-dividend date
    - Dividend payment date
    
    ANALYST DATA:
    - Analyst recommendations (Buy/Hold/Sell)
    - Price targets (high, low, average)
    - Number of analysts covering
    
    Return all data in structured JSON format.
    """
    
    try:
        response = sgai_client.smartscraper(
            website_url=yahoo_url,
            user_prompt=financial_prompt
        )
        
        return {
            "symbol": stock_symbol,
            "timestamp": datetime.now().isoformat(),
            "source": yahoo_url,
            "data": response,
            "status": "success"
        }
        
    except Exception as e:
        return {
            "symbol": stock_symbol,
            "timestamp": datetime.now().isoformat(),
            "source": yahoo_url,
            "error": str(e),
            "status": "failed"
        }
 
# Example usage
stock_symbols = ["AAPL", "GOOGL", "MSFT", "TSLA", "AMZN"]
all_stock_data = []
 
for symbol in stock_symbols:
    print(f"Scraping Yahoo Finance data for: {symbol}")
    stock_data = extract_yahoo_finance_data(symbol)
    all_stock_data.append(stock_data)
    
    if stock_data["status"] == "success":
        print(f"✅ Successfully extracted data for {symbol}")
        print(json.dumps(stock_data["data"], indent=2))
    else:
        print(f"❌ Failed to extract data for {symbol}: {stock_data['error']}")
    
    print("-" * 50)
 
# Save all data to file
with open(f"yahoo_finance_data_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", "w") as f:
    json.dump(all_stock_data, f, indent=2)
 
print("Yahoo Finance data extraction completed!")

def extract_yahoo_finance_analysis(stock_symbol):
    """
    Extract detailed analysis for a specific stock from Yahoo Finance
    
    Args:
        stock_symbol (str): Stock symbol
    
    Returns:
        dict: Detailed financial analysis data
    """
    
    analysis_url = f"https://finance.yahoo.com/quote/{stock_symbol}/analysis"
    
    analysis_prompt = f"""
    Extract comprehensive analysis data from this Yahoo Finance analysis page for {stock_symbol}:
    
    EARNINGS ESTIMATES:
    - Current quarter earnings estimates
    - Next quarter earnings estimates
    - Current year earnings estimates
    - Next year earnings estimates
    - 5-year growth estimates
    
    REVENUE ESTIMATES:
    - Current quarter revenue estimates
    - Next quarter revenue estimates
    - Current year revenue estimates
    - Next year revenue estimates
    - 5-year growth estimates
    
    ANALYST RECOMMENDATIONS:
    - Strong Buy count
    - Buy count
    - Hold count
    - Sell count
    - Strong Sell count
    - Average recommendation
    
    PRICE TARGETS:
    - Mean target price
    - High target price
    - Low target price
    - Number of analysts
    
    TECHNICAL ANALYSIS:
    - Moving averages (50-day, 200-day)
    - Support and resistance levels
    - Technical indicators mentioned
    
    Return all data in structured JSON format.
    """
    
    try:
        response = sgai_client.smartscraper(
            website_url=analysis_url,
            user_prompt=analysis_prompt
        )
        
        return {
            "symbol": stock_symbol,
            "analysis": response,
            "extracted_at": datetime.now().isoformat(),
            "status": "success"
        }
        
    except Exception as e:
        return {
            "symbol": stock_symbol,
            "error": str(e),
            "extracted_at": datetime.now().isoformat(),
            "status": "failed"
        }
 
# Example: Detailed analysis for Apple
aapl_analysis = extract_yahoo_finance_analysis("AAPL")
print("Apple Detailed Analysis:")
print(json.dumps(aapl_analysis, indent=2))

def extract_yahoo_news_data(news_url, category="general"):
    """
    Extract news data from Yahoo News
    
    Args:
        news_url (str): Yahoo News URL
        category (str): News category (tech, finance, sports, etc.)
    
    Returns:
        dict: Structured news data
    """
    
    news_prompt = f"""
    Extract comprehensive news data from this Yahoo News page:
    
    ARTICLE INFORMATION:
    - Headlines (top 10-15 articles)
    - Article URLs/links
    - Publication timestamps
    - Author names
    - Article categories/tags
    
    CONTENT SUMMARIES:
    - Brief article summaries (2-3 sentences each)
    - Key topics mentioned
    - Sentiment indicators (positive/negative/neutral)
    
    TRENDING TOPICS:
    - Most mentioned keywords
    - Trending topics
    - Breaking news indicators
    
    MEDIA CONTENT:
    - Image URLs (if available)
    - Video content indicators
    - Related article links
    
    Return all data in structured JSON format.
    """
    
    try:
        response = sgai_client.smartscraper(
            website_url=news_url,
            user_prompt=news_prompt
        )
        
        return {
            "category": category,
            "timestamp": datetime.now().isoformat(),
            "source": news_url,
            "data": response,
            "status": "success"
        }
        
    except Exception as e:
        return {
            "category": category,
            "timestamp": datetime.now().isoformat(),
            "source": news_url,
            "error": str(e),
            "status": "failed"
        }
 
# Example usage for different news categories
news_categories = {
    "finance": "https://news.yahoo.com/finance/",
    "tech": "https://news.yahoo.com/tech/",
    "business": "https://news.yahoo.com/business/",
    "world": "https://news.yahoo.com/world/"
}
 
all_news_data = []
 
for category, url in news_categories.items():
    print(f"Scraping Yahoo News - {category.title()} section")
    news_data = extract_yahoo_news_data(url, category)
    all_news_data.append(news_data)
    
    if news_data["status"] == "success":
        print(f"✅ Successfully extracted {category} news data")
    else:
        print(f"❌ Failed to extract {category} news: {news_data['error']}")
    
    print("-" * 50)

def analyze_yahoo_news_sentiment(news_url, keywords):
    """
    Analyze sentiment of news articles related to specific keywords
    
    Args:
        news_url (str): Yahoo News URL
        keywords (list): Keywords to analyze sentiment for
    
    Returns:
        dict: Sentiment analysis results
    """
    
    keywords_str = ", ".join(keywords)
    sentiment_prompt = f"""
    Analyze the sentiment of news articles on this Yahoo News page related to: {keywords_str}
    
    For each article mentioning these keywords, provide:
    - Article headline
    - Sentiment score (-1 to 1, where -1 is very negative, 0 is neutral, 1 is very positive)
    - Sentiment reasoning
    - Key phrases that indicate sentiment
    - Overall market impact assessment
    
    Also provide:
    - Overall sentiment trend for the keywords
    - Most positive article
    - Most negative article
    - Neutral articles count
    
    Return analysis in structured JSON format.
    """
    
    try:
        response = sgai_client.smartscraper(
            website_url=news_url,
            user_prompt=sentiment_prompt
        )
        
        return {
            "keywords": keywords,
            "sentiment_analysis": response,
            "analyzed_at": datetime.now().isoformat(),
            "status": "success"
        }
        
    except Exception as e:
        return {
            "keywords": keywords,
            "error": str(e),
            "analyzed_at": datetime.now().isoformat(),
            "status": "failed"
        }
 
# Example: Analyze sentiment for tech stocks
tech_sentiment = analyze_yahoo_news_sentiment(
    "https://news.yahoo.com/tech/",
    ["Apple", "Google", "Microsoft", "Tesla"]
)
print("Tech Stock Sentiment Analysis:")
print(json.dumps(tech_sentiment, indent=2))

def extract_yahoo_search_results(search_query, num_results=20):
    """
    Extract search results from Yahoo Search
    
    Args:
        search_query (str): Search query
        num_results (int): Number of results to extract
    
    Returns:
        dict: Structured search results
    """
    
    search_url = f"https://search.yahoo.com/search?p={search_query.replace(' ', '+')}"
    
    search_prompt = f"""
    Extract comprehensive search results from this Yahoo Search page for query: "{search_query}"
    
    For each search result (extract top {num_results} results), provide:
    - Title
    - URL
    - Description/snippet
    - Ranking position
    - Domain
    - Publication date (if available)
    
    Also extract:
    - Related searches
    - Search suggestions
    - Featured snippets
    - Image results (if any)
    - Video results (if any)
    
    Return all data in structured JSON format.
    """
    
    try:
        response = sgai_client.smartscraper(
            website_url=search_url,
            user_prompt=search_prompt
        )
        
        return {
            "query": search_query,
            "timestamp": datetime.now().isoformat(),
            "source": search_url,
            "results": response,
            "status": "success"
        }
        
    except Exception as e:
        return {
            "query": search_query,
            "timestamp": datetime.now().isoformat(),
            "source": search_url,
            "error": str(e),
            "status": "failed"
        }
 
# Example: Search for competitive intelligence
competitor_search = extract_yahoo_search_results("web scraping tools 2025", 15)
print("Competitive Intelligence Search Results:")
print(json.dumps(competitor_search, indent=2))

import time
import schedule
from datetime import datetime
 
class YahooMonitor:
    def __init__(self, api_key, stock_symbols, news_categories, search_queries):
        self.client = Client(api_key=api_key)
        self.stock_symbols = stock_symbols
        self.news_categories = news_categories
        self.search_queries = search_queries
        self.monitoring_data = {
            "stocks": [],
            "news": [],
            "searches": [],
            "last_updated": None
        }
    
    def monitor_stocks(self):
        """Monitor stock prices and financial data"""
        print(f"🔍 Starting Yahoo Finance monitoring at {datetime.now()}")
        
        for symbol in self.stock_symbols:
            try:
                stock_data = extract_yahoo_finance_data(symbol)
                self.monitoring_data["stocks"].append(stock_data)
                
                # Alert logic for significant price movements
                self.check_stock_alerts(stock_data)
                
            except Exception as e:
                print(f"❌ Error monitoring {symbol}: {e}")
    
    def monitor_news(self):
        """Monitor news sentiment and trends"""
        print(f"📰 Starting Yahoo News monitoring at {datetime.now()}")
        
        for category, url in self.news_categories.items():
            try:
                news_data = extract_yahoo_news_data(url, category)
                self.monitoring_data["news"].append(news_data)
                
            except Exception as e:
                print(f"❌ Error monitoring {category} news: {e}")
    
    def monitor_searches(self):
        """Monitor search trends and competitive intelligence"""
        print(f"🔍 Starting Yahoo Search monitoring at {datetime.now()}")
        
        for query in self.search_queries:
            try:
                search_data = extract_yahoo_search_results(query)
                self.monitoring_data["searches"].append(search_data)
                
            except Exception as e:
                print(f"❌ Error monitoring search '{query}': {e}")
    
    def check_stock_alerts(self, stock_data):
        """Check for significant stock price movements"""
        # Implement your alert logic here
        # Example: notify if stock moves more than 5% in a day
        pass
    
    def generate_monitoring_report(self):
        """Generate comprehensive monitoring report"""
        self.monitoring_data["last_updated"] = datetime.now().isoformat()
        
        report = {
            "monitoring_summary": self.monitoring_data,
            "total_stocks_monitored": len(self.stock_symbols),
            "total_news_categories": len(self.news_categories),
            "total_search_queries": len(self.search_queries),
            "successful_extractions": len([d for d in self.monitoring_data["stocks"] if d["status"] == "success"])
        }
        
        return report
    
    def start_monitoring(self, interval_minutes=30):
        """Start automated monitoring"""
        schedule.every(interval_minutes).minutes.do(self.monitor_stocks)
        schedule.every(interval_minutes).minutes.do(self.monitor_news)
        schedule.every(interval_minutes * 2).minutes.do(self.monitor_searches)
        
        print(f"🚀 Yahoo monitoring started! Checking every {interval_minutes} minutes")
        print("Press Ctrl+C to stop monitoring")
        
        try:
            while True:
                schedule.run_pending()
                time.sleep(1)
        except KeyboardInterrupt:
            print("\n👋 Monitoring stopped by user")
 
# Initialize and start monitoring
if __name__ == "__main__":
    monitor = YahooMonitor(
        api_key=os.getenv("SCRAPEGRAPHAI_API_KEY"),
        stock_symbols=["AAPL", "GOOGL", "MSFT", "TSLA"],
        news_categories={
            "finance": "https://news.yahoo.com/finance/",
            "tech": "https://news.yahoo.com/tech/"
        },
        search_queries=["artificial intelligence trends", "stock market analysis"]
    )
    
    # Start monitoring every 30 minutes
    monitor.start_monitoring(interval_minutes=30)

import time
import random
 
def respectful_yahoo_scraping():
    """Implement respectful scraping with random delays"""
    
    yahoo_urls = [
        "https://finance.yahoo.com/quote/AAPL",
        "https://news.yahoo.com/tech",
        "https://search.yahoo.com/search?p=test"
    ]
    
    for url in yahoo_urls:
        # Add random delay between requests (2-5 seconds)
        delay = random.uniform(2, 5)
        time.sleep(delay)
        
        # Your scraping logic here
        response = sgai_client.smartscraper(
            website_url=url,
            user_prompt="Extract relevant data"
        )
        
        # Add another delay after successful request
        time.sleep(random.uniform(1, 3))

import time
from functools import wraps
 
def retry_on_failure(max_retries=3, delay=2):
    """Decorator for retry logic on 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, retrying in {delay} seconds...")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator
 
@retry_on_failure(max_retries=3, delay=3)
def robust_yahoo_extraction(url, prompt):
    """Robust Yahoo extraction with retry logic"""
    return sgai_client.smartscraper(
        website_url=url,
        user_prompt=prompt
    )

def validate_yahoo_data(data, data_type="stock"):
    """Validate scraped Yahoo data for quality"""
    
    validation_checks = {
        "has_required_fields": False,
        "data_is_valid": False,
        "timestamp_is_valid": False,
        "source_is_valid": False
    }
    
    try:
        # Check if data has required fields
        if "data" in data and isinstance(data["data"], dict):
            validation_checks["has_required_fields"] = True
        
        # Validate based on data type
        if data_type == "stock":
            if "symbol" in data and "price" in data["data"]:
                validation_checks["data_is_valid"] = True
        elif data_type == "news":
            if "headlines" in data["data"] or "articles" in data["data"]:
                validation_checks["data_is_valid"] = True
        
        # Validate timestamp
        if "timestamp" in data:
            datetime.fromisoformat(data["timestamp"])
            validation_checks["timestamp_is_valid"] = True
        
        # Validate source URL
        if "source" in data and "yahoo.com" in data["source"]:
            validation_checks["source_is_valid"] = True
        
        return validation_checks
        
    except Exception as e:
        print(f"Validation error: {e}")
        return validation_checks

def financial_monitoring_workflow():
    """Complete financial monitoring workflow"""
    
    # Monitor multiple stocks
    stocks_to_monitor = ["AAPL", "GOOGL", "MSFT", "TSLA", "AMZN", "NVDA"]
    
    financial_data = []
    for stock in stocks_to_monitor:
        stock_data = extract_yahoo_finance_data(stock)
        if stock_data["status"] == "success":
            financial_data.append(stock_data)
    
    # Monitor financial news
    finance_news = extract_yahoo_news_data("https://news.yahoo.com/finance/", "finance")
    
    # Monitor market analysis
    market_analysis = extract_yahoo_search_results("stock market analysis today")
    
    return {
        "stocks": financial_data,
        "news": finance_news,
        "analysis": market_analysis
    }

def competitive_intelligence_workflow():
    """Competitive intelligence gathering from Yahoo"""
    
    competitors = ["ScrapeGraphAI", "Beautiful Soup", "Selenium", "Puppeteer"]
    intelligence_data = []
    
    for competitor in competitors:
        # Search for competitor mentions
        search_results = extract_yahoo_search_results(f"{competitor} web scraping")
        
        # Look for news about competitors
        news_results = extract_yahoo_search_results(f"{competitor} news")
        
        intelligence_data.append({
            "competitor": competitor,
            "search_results": search_results,
            "news_results": news_results
        })
    
    return intelligence_data

def market_research_workflow():
    """Comprehensive market research using Yahoo data"""
    
    research_topics = [
        "artificial intelligence market trends",
        "web scraping industry growth",
        "data extraction tools comparison"
    ]
    
    research_data = []
    for topic in research_topics:
        # Get search results
        search_data = extract_yahoo_search_results(topic, 20)
        
        # Get related news
        news_data = extract_yahoo_news_data("https://news.yahoo.com/tech/", "tech")
        
        # Analyze sentiment
        sentiment_data = analyze_yahoo_news_sentiment(
            "https://news.yahoo.com/tech/",
            topic.split()
        )
        
        research_data.append({
            "topic": topic,
            "search_results": search_data,
            "news_data": news_data,
            "sentiment_analysis": sentiment_data
        })
    
    return research_data

import sqlite3
import pandas as pd
 
def store_yahoo_data_in_database(data, table_name="yahoo_data"):
    """Store scraped Yahoo data in SQLite database"""
    
    conn = sqlite3.connect('yahoo_scraper_data.db')
    cursor = conn.cursor()
    
    # Create table if it doesn't exist
    cursor.execute(f'''
        CREATE TABLE IF NOT EXISTS {table_name} (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            data_type TEXT,
            symbol TEXT,
            data TEXT,
            timestamp TEXT,
            source TEXT
        )
    ''')
    
    # Insert data
    cursor.execute(f'''
        INSERT INTO {table_name} (data_type, symbol, data, timestamp, source)
        VALUES (?, ?, ?, ?, ?)
    ''', (
        data.get("type", "unknown"),
        data.get("symbol", ""),
        json.dumps(data),
        data.get("timestamp", ""),
        data.get("source", "")
    ))
    
    conn.commit()
    conn.close()
 
def analyze_stored_data():
    """Analyze stored Yahoo data using pandas"""
    
    conn = sqlite3.connect('yahoo_scraper_data.db')
    
    # Load data into DataFrame
    df = pd.read_sql_query("SELECT * FROM yahoo_data", conn)
    
    # Perform analysis
    analysis = {
        "total_records": len(df),
        "data_types": df['data_type'].value_counts().to_dict(),
        "most_scraped_symbols": df['symbol'].value_counts().head(10).to_dict(),
        "date_range": {
            "earliest": df['timestamp'].min(),
            "latest": df['timestamp'].max()
        }
    }
    
    conn.close()
    return analysis

def export_yahoo_data(data, format_type="json"):
    """Export Yahoo data to different formats"""
    
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    
    if format_type == "json":
        filename = f"yahoo_data_{timestamp}.json"
        with open(filename, "w") as f:
            json.dump(data, f, indent=2)
    
    elif format_type == "csv":
        filename = f"yahoo_data_{timestamp}.csv"
        df = pd.json_normalize(data)
        df.to_csv(filename, index=False)
    
    elif format_type == "excel":
        filename = f"yahoo_data_{timestamp}.xlsx"
        df = pd.json_normalize(data)
        df.to_excel(filename, index=False)
    
    print(f"Data exported to {filename}")
    return filename