fix data downloader

portfolio_calculator.py

@@ -15,7 +15,14 @@ import numpy as np
 import pandas as pd
 import yfinance as yf
 import streamlit as st
-import time
+from concurrent.futures import ProcessPoolExecutor, as_completed
+import logging
+from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
+from ratelimit import limits, sleep_and_retry
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
 
 # Constants
@@ -24,6 +31,92 @@ MIN_DATA_POINTS = 30
 MAX_TICKERS = 20
 
 
+# Rate limiter: Max 5 calls per 10 seconds per ticker (conservative for Yahoo Finance)
+@sleep_and_retry
+@limits(calls=5, period=10)
+@retry(
+    stop=stop_after_attempt(3),  # Max 3 retries
+    wait=wait_exponential(multiplier=1, min=2, max=60),  # Exponential backoff: 2s, 4s, 8s, ... max 60s
+    retry=retry_if_exception_type((ConnectionError, TimeoutError)),  # Only retry on network errors
+    reraise=True
+)
+def _fetch_with_retry(ticker_obj, method: str, period: str = "1y"):
+    """
+    Internal function with retry logic for fetching data.
+
+    Args:
+        ticker_obj: yfinance Ticker object
+        method: Which method to use ('period', 'short', 'dates')
+        period: Time period
+
+    Returns:
+        Historical data or None
+    """
+    if method == 'period':
+        return ticker_obj.history(period=period)
+    elif method == 'short':
+        return ticker_obj.history(period="6mo")
+    else:  # dates
+        import datetime
+        end_date = datetime.datetime.now()
+        start_date = end_date - datetime.timedelta(days=365)
+        return ticker_obj.history(start=start_date, end=end_date)
+
+
+def fetch_single_ticker(ticker: str, period: str = "1y") -> Tuple[str, Optional[pd.Series], Optional[str]]:
+    """
+    Fetch historical data for a single ticker with rate limiting and exponential backoff.
+
+    This function runs in a separate process for parallel execution.
+    Uses tenacity for exponential backoff and ratelimit for request throttling.
+
+    Args:
+        ticker: Ticker symbol
+        period: Time period for historical data
+
+    Returns:
+        Tuple of (ticker, price_series, error_message)
+    """
+    try:
+        # Create fresh Ticker object in this process
+        ticker_obj = yf.Ticker(ticker)
+
+        # Try Method 1: Standard period with retry logic
+        try:
+            hist = _fetch_with_retry(ticker_obj, 'period', period)
+            if not hist.empty:
+                logger.info(f"✅ {ticker}: Fetched {len(hist)} days (method: period)")
+                return ticker, hist['Close'], None
+        except Exception as e:
+            logger.warning(f"⚠️ {ticker}: Method 1 failed - {str(e)}")
+
+        # Try Method 2: Shorter period
+        try:
+            hist = _fetch_with_retry(ticker_obj, 'short', period)
+            if not hist.empty:
+                logger.info(f"✅ {ticker}: Fetched {len(hist)} days (method: short)")
+                return ticker, hist['Close'], None
+        except Exception as e:
+            logger.warning(f"⚠️ {ticker}: Method 2 failed - {str(e)}")
+
+        # Try Method 3: Explicit dates (most reliable)
+        try:
+            hist = _fetch_with_retry(ticker_obj, 'dates', period)
+            if not hist.empty:
+                logger.info(f"✅ {ticker}: Fetched {len(hist)} days (method: dates)")
+                return ticker, hist['Close'], None
+        except Exception as e:
+            logger.warning(f"⚠️ {ticker}: Method 3 failed - {str(e)}")
+
+        # All methods failed
+        logger.error(f"❌ {ticker}: All methods exhausted")
+        return ticker, None, "No data available after all retry attempts"
+
+    except Exception as e:
+        logger.error(f"❌ {ticker}: Fatal error - {str(e)}")
+        return ticker, None, str(e)
+
+
 @st.cache_data(ttl=3600)  # Cache for 1 hour
 def fetch_historical_data(
     tickers: Tuple[str, ...],  # Tuple for hashability (caching requirement)
@@ -73,60 +166,50 @@ def fetch_historical_data(
 
     except Exception as e:
         # Log the error but continue to fallback strategy
-        print(f"Batch download failed: {str(e)}, trying individual downloads...")
+        logger.warning(f"Batch download failed: {str(e)}, trying individual downloads...")
+
+    # Strategy 2: Parallel download using ProcessPoolExecutor
+    st.info(f"📥 Fetching data for {len(ticker_list)} tickers in parallel...")
 
-    # Strategy 2: Download one ticker at a time with retries
-    st.info("Fetching data individually for each ticker (this may take a moment)...")
     individual_prices = {}
     failed_tickers = []
 
-    for idx, ticker in enumerate(ticker_list):
-        max_retries = 3
-        retry_delay = 2  # seconds
-
-        for attempt in range(max_retries):
-            try:
-                # Add delay between requests to avoid rate limiting (except first ticker)
-                if idx > 0 or attempt > 0:
-                    time.sleep(retry_delay)
-
-                # Use Ticker object for more reliable downloads
-                ticker_obj = yf.Ticker(ticker)
-
-                # Try different methods
-                if attempt == 0:
-                    # Method 1: history with period
-                    hist = ticker_obj.history(period=period)
-                elif attempt == 1:
-                    # Method 2: Try shorter period
-                    hist = ticker_obj.history(period="6mo")
-                else:
-                    # Method 3: Try start/end dates
-                    import datetime
-                    end_date = datetime.datetime.now()
-                    start_date = end_date - datetime.timedelta(days=365)
-                    hist = ticker_obj.history(start=start_date, end=end_date)
-
-                if not hist.empty:
-                    # Extract close prices
-                    individual_prices[ticker] = hist['Close']
-                    st.success(f"✅ {ticker}: {len(hist)} days of data")
-                    break  # Success, exit retry loop
-                else:
-                    if attempt < max_retries - 1:
-                        st.warning(f"⚠️ {ticker}: No data, retrying... (attempt {attempt + 1}/{max_retries})")
+    # Determine number of workers (max 4 to avoid overwhelming the API)
+    max_workers = min(len(ticker_list), 4)
+
+    try:
+        # Use ProcessPoolExecutor for true parallel execution
+        with ProcessPoolExecutor(max_workers=max_workers) as executor:
+            # Submit all ticker fetch jobs
+            future_to_ticker = {
+                executor.submit(fetch_single_ticker, ticker, period): ticker
+                for ticker in ticker_list
+            }
+
+            # Process results as they complete
+            completed = 0
+            for future in as_completed(future_to_ticker):
+                ticker = future_to_ticker[future]
+                completed += 1
+
+                try:
+                    ticker_symbol, price_series, error = future.result(timeout=30)
+
+                    if price_series is not None and not price_series.empty:
+                        individual_prices[ticker_symbol] = price_series
+                        st.success(f"✅ {ticker_symbol}: {len(price_series)} days ({completed}/{len(ticker_list)})")
                     else:
-                        st.error(f"❌ {ticker}: No data after {max_retries} attempts")
-                        failed_tickers.append(ticker)
-
-            except Exception as e:
-                if attempt < max_retries - 1:
-                    st.warning(f"⚠️ {ticker}: Error, retrying... (attempt {attempt + 1}/{max_retries})")
-                    retry_delay *= 2  # Exponential backoff
-                else:
+                        failed_tickers.append(ticker_symbol)
+                        st.error(f"❌ {ticker_symbol}: {error or 'No data'} ({completed}/{len(ticker_list)})")
+
+                except Exception as e:
                     failed_tickers.append(ticker)
-                    st.error(f"❌ {ticker}: Failed after {max_retries} attempts")
-                    print(f"Failed to fetch {ticker}: {str(e)}")
+                    st.error(f"❌ {ticker}: {str(e)} ({completed}/{len(ticker_list)})")
+
+    except Exception as e:
+        st.error(f"Parallel processing error: {str(e)}")
+        # Fall back to empty result
+        pass
 
     # Check if we got any data
     if not individual_prices:

requirements.txt

@@ -6,3 +6,5 @@ pandas==2.2.0
 numpy==1.26.3
 sympy==1.12
 matplotlib==3.8.2
+tenacity==8.2.3
+ratelimit==2.2.1