data/historical.py

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))