From 6167bb01c5961b3d89152f00367771e67b35bbee Mon Sep 17 00:00:00 2001 From: Harwinder Date: Tue, 5 May 2026 16:46:37 -0400 Subject: [PATCH] Fix syntax error and dataframe column extraction in yfinance batch fetch --- data_provider.py | 165 +++++++++++++++++++++++++---------------------- 1 file changed, 87 insertions(+), 78 deletions(-) diff --git a/data_provider.py b/data_provider.py index 86c2a57..97a793c 100644 --- a/data_provider.py +++ b/data_provider.py @@ -212,38 +212,44 @@ class ForexDataProvider: else: period = f"{lookback_days}d" - try: - df = yf.download( - yf_symbol, interval=yf_interval, period=period, - progress=False, auto_adjust=False, threads=False, - ) - if df is None or df.empty: - raise DataUnavailableError(f"No data returned for {symbol}") + for attempt in range(3): + try: + df = yf.download( + yf_symbol, interval=yf_interval, period=period, + progress=False, auto_adjust=False, threads=False, + ) + if df is not None and not df.empty: + break + time.sleep(1.0) + except Exception as e: + if attempt == 2: + raise DataUnavailableError(f"yfinance fetch failed for {symbol} {timeframe.value}: {e}") + time.sleep(1.0) + else: + raise DataUnavailableError(f"No data returned for {symbol}") - if isinstance(df.columns, pd.MultiIndex): - df.columns = df.columns.get_level_values(0) + if isinstance(df.columns, pd.MultiIndex): + df.columns = df.columns.get_level_values(0) - if df.index.tz is None: - df.index = df.index.tz_localize("UTC") - else: - df.index = df.index.tz_convert("UTC") - df.index.name = "Datetime" - df = df[["Open", "High", "Low", "Close", "Volume"]].copy() + if df.index.tz is None: + df.index = df.index.tz_localize("UTC") + else: + df.index = df.index.tz_convert("UTC") + df.index.name = "Datetime" + df = df[["Open", "High", "Low", "Close", "Volume"]].copy() - if resample_needed and timeframe == Timeframe.H4: - # Resample 1h data into 4h blocks - df = df.resample('4h').agg({ - 'Open': 'first', - 'High': 'max', - 'Low': 'min', - 'Close': 'last', - 'Volume': 'sum' - }).dropna() + if resample_needed and timeframe == Timeframe.H4: + # Resample 1h data into 4h blocks + df = df.resample('4h').agg({ + 'Open': 'first', + 'High': 'max', + 'Low': 'min', + 'Close': 'last', + 'Volume': 'sum' + }).dropna() - logger.info(f"[yf live] {yf_symbol} {timeframe.value}: {len(df)} bars, last close={df['Close'].iloc[-1]:.5f}") - return df - except Exception as e: - raise DataUnavailableError(f"yfinance fetch failed for {symbol} {timeframe.value}: {e}") + logger.info(f"[yf live] {yf_symbol} {timeframe.value}: {len(df)} bars, last close={df['Close'].iloc[-1]:.5f}") + return df def _batch_fetch_timeframe(self, symbols: List[str], timeframe: Timeframe, lookback_days: int) -> None: if not _YF_AVAILABLE: @@ -270,64 +276,67 @@ class ForexDataProvider: else: period = f"{lookback_days}d" - try: - df_multi = yf.download( - " ".join(yf_symbols), interval=yf_interval, period=period, - progress=False, auto_adjust=False, threads=True, group_by="ticker" - ) - - if df_multi is None or df_multi.empty: - logger.warning(f"Batch fetch empty for {timeframe.value}") - return + for attempt in range(3): + try: + df_multi = yf.download( + " ".join(yf_symbols), interval=yf_interval, period=period, + progress=False, auto_adjust=False, threads=False, group_by="ticker" + ) + if df_multi is not None and not df_multi.empty: + break + time.sleep(1.0) + except Exception as e: + logger.warning(f"Batch fetch attempt {attempt+1} failed for {timeframe.value}: {e}") + time.sleep(1.0) + else: + logger.warning(f"Batch fetch empty for {timeframe.value} after 3 attempts") + return - if df_multi.index.tz is None: - df_multi.index = df_multi.index.tz_localize("UTC") - else: - df_multi.index = df_multi.index.tz_convert("UTC") - df_multi.index.name = "Datetime" + if df_multi.index.tz is None: + df_multi.index = df_multi.index.tz_localize("UTC") + else: + df_multi.index = df_multi.index.tz_convert("UTC") + df_multi.index.name = "Datetime" - now = time.time() - for symbol, yf_sym in zip(symbols, yf_symbols): - try: - if len(yf_symbols) == 1: - df_sym = df_multi.copy() - if isinstance(df_sym.columns, pd.MultiIndex): - df_sym.columns = df_sym.columns.get_level_values(0) + now = time.time() + for symbol, yf_sym in zip(symbols, yf_symbols): + try: + if len(yf_symbols) == 1: + df_sym = df_multi.copy() + if isinstance(df_sym.columns, pd.MultiIndex): + df_sym.columns = df_sym.columns.get_level_values(1) + else: + if yf_sym in df_multi.columns.get_level_values(0): + df_sym = df_multi[yf_sym].copy() else: - if yf_sym in df_multi.columns.get_level_values(0): - df_sym = df_multi[yf_sym].copy() + # Might be single level columns if download only returned 1 valid ticker + if yf_sym == yf_symbols[0] and not isinstance(df_multi.columns, pd.MultiIndex): + df_sym = df_multi.copy() else: - # Might be single level columns if download only returned 1 valid ticker - if yf_sym == yf_symbols[0] and not isinstance(df_multi.columns, pd.MultiIndex): - df_sym = df_multi.copy() - else: - continue + continue - # Ignore if column names are not Open/High/Low/Close - if 'Close' not in df_sym.columns: - continue + # Ignore if column names are not Open/High/Low/Close + if 'Close' not in df_sym.columns: + continue - df_sym = df_sym[["Open", "High", "Low", "Close", "Volume"]].copy() + df_sym = df_sym[["Open", "High", "Low", "Close", "Volume"]].copy() - if resample_needed and timeframe == Timeframe.H4: - df_sym = df_sym.resample('4h').agg({ - 'Open': 'first', - 'High': 'max', - 'Low': 'min', - 'Close': 'last', - 'Volume': 'sum' - }).dropna() + if resample_needed and timeframe == Timeframe.H4: + df_sym = df_sym.resample('4h').agg({ + 'Open': 'first', + 'High': 'max', + 'Low': 'min', + 'Close': 'last', + 'Volume': 'sum' + }).dropna() - df_sym = self._validate_and_clean(df_sym, symbol, timeframe) - cache_key = f"{symbol}_{timeframe.value}" - self._cache[cache_key] = df_sym - self._cache_timestamps[cache_key] = now - logger.info(f"[yf batch] {yf_sym} {timeframe.value}: {len(df_sym)} bars, last close={df_sym['Close'].iloc[-1]:.5f}") - except Exception as e: - logger.warning(f"Failed to process batch data for {symbol}: {e}") - - except Exception as e: - logger.warning(f"Batch fetch failed for {timeframe.value}: {e}") + df_sym = self._validate_and_clean(df_sym, symbol, timeframe) + cache_key = f"{symbol}_{timeframe.value}" + self._cache[cache_key] = df_sym + self._cache_timestamps[cache_key] = now + logger.info(f"[yf batch] {yf_sym} {timeframe.value}: {len(df_sym)} bars, last close={df_sym['Close'].iloc[-1]:.5f}") + except Exception as e: + logger.warning(f"Failed to process batch data for {symbol}: {e}") def _validate_and_clean( self, df: pd.DataFrame, symbol: str, timeframe: Timeframe