import random import time from datetime import datetime, timedelta, timezone import pandas as pd from config import YF_NATIVE_INTERVALS, CUSTOM_TIMEFRAMES from features.feature_pipeline import compute_feature_frame_for_period from utils.helpers import feature_warmup_days from data.validation import DataFetchError, validate_ohlcv, positive_integer _YF_TIMEOUT_SECONDS = 30 _RETENTION_GUARD_SECONDS = 300 OHLCV_COLUMNS = ['Open', 'High', 'Low', 'Close', 'Volume'] def _fetch_yf(symbol: str, interval: str, start: datetime, end: datetime) -> pd.DataFrame: import yfinance as yf last_error = None for attempt in range(3): try: return yf.Ticker(symbol).history( start=start, end=end, interval=interval, auto_adjust=False, actions=False, timeout=_YF_TIMEOUT_SECONDS, raise_errors=True, ) except Exception as error: last_error = error if attempt < 2: time.sleep(0.5 * (attempt + 1) + random.uniform(0.0, 0.25)) raise DataFetchError(f"Yahoo Finance could not fetch {symbol} ({interval}) after 3 attempts ({type(last_error).__name__}).") from None def _download_window(symbol: str, interval: str, start: datetime, end: datetime) -> pd.DataFrame: df = _fetch_yf(symbol, interval, start, end) if df is None or df.empty: return pd.DataFrame() if isinstance(df.columns, pd.MultiIndex): # yfinance can return either (field, ticker) or (ticker, field), # depending on version and whether the ticker is cached. fields = set(OHLCV_COLUMNS).intersection(set(df.columns.get_level_values(0))) if fields: df = df.droplevel(1, axis=1) else: fields = set(OHLCV_COLUMNS).intersection(set(df.columns.get_level_values(-1))) if fields: df = df.droplevel(0, axis=1) df = df.rename(columns=lambda value: str(value).title()) df.index.name = 'Datetime' keep = [c for c in OHLCV_COLUMNS if c in df.columns] return df[keep] def _finalize_ohlcv(df: pd.DataFrame) -> pd.DataFrame: if df is None or df.empty: return df return validate_ohlcv(df, normalize=True) def _native_fetch(symbol: str, interval: str, days: int) -> pd.DataFrame: days = positive_integer(days, 'lookback_days', 10000) if interval not in YF_NATIVE_INTERVALS: raise DataFetchError(f"'{interval}' is not a native yfinance interval.") limits = YF_NATIVE_INTERVALS[interval] max_days, request_days = (limits['max_days'], limits['max_request_days']) if max_days is not None: days = min(days, max_days) if request_days is None: request_days = days frames, empty_windows = [], 0 window_end = datetime.now(timezone.utc) # Keep a full-retention request inside Yahoo's rolling boundary without # shortening the requested interval. Moving the exclusive end slightly # into the future is safe because the provider returns only available # candles, while it preserves the exact span used for chunk pagination. if max_days is not None and days == max_days: window_end += timedelta(seconds=_RETENTION_GUARD_SECONDS) remaining = days while remaining > 0: window_days = min(request_days, remaining) window_start = window_end - timedelta(days=window_days) frame = _download_window(symbol, interval, window_start, window_end) if not frame.empty: frames.append(frame) else: empty_windows += 1 window_end = window_start remaining -= window_days if remaining > 0: time.sleep(0.1 + random.uniform(0, 0.2)) df = _finalize_ohlcv(pd.concat(frames)) if frames else pd.DataFrame() if df.empty or 'Close' not in df.columns: limit_note = f" Yahoo keeps at most {max_days} days of '{interval}' history." if max_days is not None else '' raise DataFetchError(f"No usable data returned for '{symbol}' at interval '{interval}' over the last {days} day(s). Check it's a valid yfinance ticker (e.g. EURUSD=X, BTC-USD) and that the market has traded recently." + limit_note) df.attrs['effective_lookback_days'] = days df.attrs['source_interval'] = interval df.attrs['empty_request_windows'] = empty_windows df.attrs['retention_guard_seconds'] = _RETENTION_GUARD_SECONDS if max_days is not None and days == max_days else 0 df.attrs['requested_earliest_start_utc'] = window_start.isoformat() return df def resample_ohlc(df: pd.DataFrame, rule: str, source_step=None) -> pd.DataFrame: if df.empty: return df.copy() if not isinstance(df.index, pd.DatetimeIndex) or df.index.isna().any() or df.index.duplicated().any() or not df.index.is_monotonic_increasing: raise DataFetchError('Resampling needs ordered, unique datetime candles.') agg = {'Open': 'first', 'High': 'max', 'Low': 'min', 'Close': 'last', 'Volume': 'sum'} agg = {k: v for k, v in agg.items() if k in df.columns} price_cols = [c for c in ('Open', 'High', 'Low', 'Close') if c in agg] grouped = df.resample(rule, origin='epoch', label='left', closed='left') result = grouped.agg(agg).dropna(subset=price_cols) if source_step is not None: source_step = pd.Timedelta(source_step) target_step = pd.Timedelta(rule) if source_step <= pd.Timedelta(0) or target_step < source_step or target_step % source_step != pd.Timedelta(0): raise ValueError('Resampling requires a positive integral multiple of the source interval.') expected = int(target_step / source_step) counts = grouped['Close'].count() times = pd.Series(df.index, index=df.index) first = times.resample(rule, origin='epoch').first() last = times.resample(rule, origin='epoch').last() # Count + endpoints alone do not prove that interior source candles are # on-grid (two shifted candles could previously form an invented bar). on_grid = pd.Series((df.index.asi8 % source_step.value) == 0, index=df.index) grid_ok = on_grid.astype('int8').resample(rule, origin='epoch').min().eq(1) complete = (counts == expected) & grid_ok & (first == counts.index) & (last == counts.index + target_step - source_step) result = result.loc[complete.reindex(result.index, fill_value=False)] result.attrs.update(df.attrs) return result def get_historical(symbol: str, timeframe: str, lookback_days: int=None) -> pd.DataFrame: if not symbol or not symbol.strip(): raise DataFetchError('No symbol given.') symbol = symbol.strip() if timeframe in CUSTOM_TIMEFRAMES: spec = CUSTOM_TIMEFRAMES[timeframe] source_interval, rule = (spec['source'], spec['rule']) elif timeframe in YF_NATIVE_INTERVALS: source_interval, rule = (timeframe, None) else: raise DataFetchError(f"Unknown timeframe '{timeframe}'. Valid options: {', '.join(CUSTOM_TIMEFRAMES.keys())}") max_days = YF_NATIVE_INTERVALS[source_interval]['max_days'] requested = positive_integer(lookback_days, 'lookback_days', 10000) if lookback_days is not None else 0 days = requested if requested > 0 else max_days if max_days else 730 df = _native_fetch(symbol, source_interval, days) if rule: source_minutes = 60 if source_interval == '60m' else int(source_interval.rstrip('m')) df = resample_ohlc(df, rule, source_step=pd.Timedelta(minutes=source_minutes)) if df.empty: raise DataFetchError(f"Not enough '{source_interval}' candles to build '{timeframe}' bars. Try a longer lookback.") return df def get_historical_with_features(symbol: str, timeframe: str, lookback_days: int, use_features: bool): if not use_features: return (get_historical(symbol, timeframe, lookback_days=int(lookback_days)), None) padding_days = feature_warmup_days(timeframe, minimum_days=0) df_padded = get_historical(symbol, timeframe, lookback_days=int(lookback_days) + padding_days) return compute_feature_frame_for_period(df_padded, keep_days=int(lookback_days))