fetch_stock_data

fetch_func.py

def fetch_stock_data(tickers: List[str], start_date: str, end_date: str, exchanges: Dict[str, str] = None) -> Tuple[pd.DataFrame, Dict]:
    """
    Fetch stock price data and exchange information.
    
    Args:
        tickers: List of stock tickers
        start_date: Start date (YYYY-MM-DD)
        end_date: End date (YYYY-MM-DD)
        exchanges: Dictionary mapping tickers to their respective exchanges (optional)
        
    Returns:
        Tuple of price data and ticker information
    """
    try:
        # Convert date formats to YYYY-MM-DD
        try:
            start_date_str = convert_timestamp(start_date)
            end_date_str = convert_timestamp(end_date)
            print(f"Loading data for {tickers} from {start_date_str} to {end_date_str}")
        except Exception as date_error:
            print(f"Error converting date format: {date_error}")
            raise ValueError(f"Invalid date format: {date_error}")
        
        # Load data
        data = yf.download(tickers, start=start_date_str, end=end_date_str)
        
        # Check if data is empty
        if data.empty:
            raise ValueError(f"Could not load data for period {start_date_str} to {end_date_str}")
        
        # Get closing price data
        close_data = data['Close']
        
        # Check for missing tickers
        missing_tickers = []
        for ticker in tickers:
            if ticker not in close_data.columns:
                missing_tickers.append(ticker)
            elif close_data[ticker].isna().all():
                missing_tickers.append(ticker)
        
        if missing_tickers:
            print(f"Warning: No data available for tickers: {missing_tickers}")
        
        # Get or use provided exchange information for each ticker
        ticker_info = {}
        for ticker in tickers:
            if exchanges and ticker in exchanges:
                # Use provided exchange information
                ticker_info[ticker] = exchanges.get(ticker, 'N/A')
            elif ticker in close_data.columns:
                # Fetch exchange information if not provided
                try:
                    info = yf.Ticker(ticker).info
                    exchange = info.get('exchange', 'N/A')
                    ticker_info[ticker] = exchange
                except Exception as e:
                    print(f"Unable to get information for {ticker}: {e}")
                    ticker_info[ticker] = 'N/A'
            else:
                ticker_info[ticker] = 'N/A'
        
        # Handle missing data if any
        if close_data.isna().any().any():
            close_data = close_data.ffill().bfill()
            print("Filled missing values in data")
        
        return close_data, ticker_info
        
    except Exception as e:
        print(f"Error loading stock data: {e}")
        # Return empty DataFrame and default information
        empty_df = pd.DataFrame(columns=tickers)
        ticker_info = {ticker: exchanges.get(ticker, 'N/A') if exchanges else 'N/A' for ticker in tickers}
        return empty_df, ticker_info