ForexBot2/data_provider.py

"""
ForexBot2 — Data Provider (Twelve Data SDK)
==========================================================
Fetches and validates OHLCV via the official `twelvedata` Python SDK.
"""

import logging
import os
import time
from collections import deque
from typing import Dict, Optional

import pandas as pd
from twelvedata import TDClient

from config import PlatformConfig, Timeframe, XAUUSD, PAIR_INFO

logger = logging.getLogger(__name__)


def _is_forex_weekend(now=None) -> bool:
    """Forex market closed: Friday 22:00 UTC → Sunday 22:00 UTC."""
    if now is None:
        now = pd.Timestamp.now(tz="UTC")
    wd = now.weekday()
    hour = now.hour
    if wd == 4 and hour >= 22:
        return True
    if wd == 5:
        return True
    if wd == 6 and hour < 22:
        return True
    return False


class DataIntegrityError(Exception):
    """Raised when data fails validation checks."""
    pass


class ForexDataProvider:
    """Twelve Data SDK client + on-disk parquet cache."""

    TD_INTERVALS = {
        Timeframe.M1:  "1min",
        Timeframe.M5:  "5min",
        Timeframe.M15: "15min",
        Timeframe.M30: "30min",
        Timeframe.H1:  "1h",
        Timeframe.H4:  "4h",
        Timeframe.D1:  "1day",
        Timeframe.W1:  "1week",
    }

    BARS_PER_DAY = {
        Timeframe.M1: 1440, Timeframe.M5: 288, Timeframe.M15: 96,
        Timeframe.M30: 48, Timeframe.H1: 24, Timeframe.H4: 6,
        Timeframe.D1: 1, Timeframe.W1: 0.143,
    }

    CANDLE_SECONDS = {
        Timeframe.M1: 60, Timeframe.M5: 300, Timeframe.M15: 900,
        Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400,
        Timeframe.D1: 86400, Timeframe.W1: 604800,
    }

    RATE_LIMIT = 55
    RATE_WINDOW = 60.0

    def __init__(self, config: PlatformConfig):
        self.config = config
        self._api_key = (
            os.environ.get("TWELVE_DATA_API_KEY")
            or config.twelve_data_api_key
        )
        self._cache: Dict[str, pd.DataFrame] = {}
        self._cache_timestamps: Dict[str, float] = {}
        self._request_times: deque = deque()
        self._td: Optional[TDClient] = (
            TDClient(apikey=self._api_key) if self._api_key else None
        )

    @staticmethod
    def _to_td_symbol(symbol: str) -> str:
        if "/" in symbol:
            return symbol
        clean = symbol.replace("=X", "")
        return f"{clean[:3]}/{clean[3:]}"

    # ── Public API ─────────────────────────────────────────────────

    def fetch_ohlcv(
        self,
        symbol: str = XAUUSD,
        timeframe: Timeframe = Timeframe.H1,
        lookback_days: Optional[int] = None,
    ) -> pd.DataFrame:
        """Fetch OHLCV with in-memory caching for live mode."""
        if lookback_days is None:
            lookback_days = self.config.data_lookback_days

        cache_key = f"{symbol}_{timeframe.value}"
        now = time.time()

        cache_ttl = self.CANDLE_SECONDS.get(timeframe, self.config.refresh_interval)
        if (
            cache_key in self._cache_timestamps
            and (now - self._cache_timestamps[cache_key]) < cache_ttl
            and cache_key in self._cache
        ):
            cached = self._cache.get(cache_key)
            if cached is not None and not cached.empty:
                if self._is_data_fresh(cached, symbol, timeframe):
                    return cached
                logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching")
                del self._cache[cache_key]
                del self._cache_timestamps[cache_key]

        outputsize = max(
            300,
            min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, 5000),
        )
        df = self._fetch_twelve_data(symbol, timeframe, outputsize)
        df = self._drop_forming_candle(df, timeframe)
        df = self._validate_and_clean(df, symbol, timeframe)

        if not self._is_data_fresh(df, symbol, timeframe):
            logger.warning(
                f"STALE DATA from API: {symbol} {timeframe.value} — "
                f"last bar {df.index[-1]} is too old. Signal quality may be degraded."
            )

        self._cache[cache_key] = df
        self._cache_timestamps[cache_key] = now
        return df

    def _drop_forming_candle(self, df: pd.DataFrame, timeframe: Timeframe) -> pd.DataFrame:
        """Drop the still-forming last candle so analysis only sees closed bars.

        Twelve Data's time_series includes the current partial candle; the
        scanner fires seconds after the boundary, so without this guard every
        indicator reads a seconds-old stub bar and signals repaint.
        """
        if df is None or df.empty:
            return df
        candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
        now = pd.Timestamp.now(tz="UTC")
        if df.index[-1] + pd.Timedelta(seconds=candle_seconds) > now:
            df = df.iloc[:-1]
        return df

    def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe) -> bool:
        """Check if the last candle is within acceptable age for this timeframe."""
        if df is None or df.empty:
            return False

        now = pd.Timestamp.now(tz="UTC")

        if _is_forex_weekend(now):
            return True

        last_bar = df.index[-1]
        candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
        max_age_seconds = candle_seconds * 3
        age_seconds = (now - last_bar).total_seconds()

        if age_seconds > max_age_seconds:
            logger.warning(
                f"{symbol} {timeframe.value}: last bar {last_bar.strftime('%Y-%m-%d %H:%M')} "
                f"is {age_seconds / 60:.0f}m old (limit: {max_age_seconds / 60:.0f}m)"
            )
            return False

        return True

    def clear_cache(self) -> int:
        """Wipe all cached data. Returns number of entries cleared."""
        count = len(self._cache)
        self._cache.clear()
        self._cache_timestamps.clear()
        logger.info(f"Cache cleared: {count} entries wiped")
        return count

    def get_latest_price(self, symbol: str = XAUUSD) -> Optional[float]:
        try:
            df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1)
            if df is not None and not df.empty:
                return float(df["Close"].iloc[-1])
        except Exception:
            pass
        return None

    def get_spread_estimate(self, symbol: str = XAUUSD) -> float:
        try:
            df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1)
            if df is not None and not df.empty:
                return float((df["High"] - df["Low"]).median())
        except Exception:
            pass
        info = PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])
        return info["pip_value"]

    def get_pair_info(self, symbol: str = XAUUSD) -> Dict:
        info = PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])
        return {"symbol": symbol, **info}

    # ── Private ────────────────────────────────────────────────────

    def _throttle(self, credits: int = 1) -> None:
        """Pro tier: 55 credits / minute. Block until budget allows."""
        while True:
            now = time.time()
            while self._request_times and now - self._request_times[0] >= self.RATE_WINDOW:
                self._request_times.popleft()

            available = self.RATE_LIMIT - len(self._request_times)
            if available >= credits:
                for _ in range(credits):
                    self._request_times.append(time.time())
                return

            wait = (self._request_times[0] + self.RATE_WINDOW) - now + 0.2
            logger.info(
                f"Rate limit: {len(self._request_times)}/{self.RATE_LIMIT} credits used "
                f"— waiting {wait:.1f}s..."
            )
            time.sleep(wait)

    def _fetch_twelve_data(
        self,
        symbol: str,
        timeframe: Timeframe,
        outputsize: int,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
    ) -> pd.DataFrame:
        if self._td is None:
            raise DataIntegrityError(
                "No Twelve Data API key. Set TWELVE_DATA_API_KEY env var "
                "or config.twelve_data_api_key."
            )

        td_symbol = self._to_td_symbol(symbol)
        interval = self.TD_INTERVALS[timeframe]

        self._throttle(credits=1)

        kwargs = {
            "symbol": td_symbol,
            "interval": interval,
            "outputsize": min(outputsize, 5000),
            "timezone": "UTC",
        }
        if start_date:
            kwargs["start_date"] = start_date
        if end_date:
            kwargs["end_date"] = end_date

        try:
            df = self._td.time_series(**kwargs).as_pandas()
        except Exception as e:
            raise DataIntegrityError(
                f"Twelve Data fetch failed for {td_symbol}: {e}"
            ) from e

        if df is None or df.empty:
            raise DataIntegrityError(f"Empty response from Twelve Data for {td_symbol}")

        # SDK returns lowercase columns; engine expects capitalised.
        df = df.rename(columns={
            "open": "Open", "high": "High", "low": "Low",
            "close": "Close", "volume": "Volume",
        })
        for col in ("Open", "High", "Low", "Close", "Volume"):
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors="coerce")
        if "Volume" not in df.columns:
            df["Volume"] = 0.0

        # SDK returns newest-first; engine + indicators want oldest-first.
        df = df.sort_index()
        if df.index.tz is None:
            df.index = df.index.tz_localize("UTC")
        df.index.name = "Datetime"

        logger.debug(f"Fetched {len(df)} bars for {td_symbol} @ {interval}")
        return df

    def _validate_and_clean(
        self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
    ) -> pd.DataFrame:
        if df is None or df.empty:
            raise DataIntegrityError(f"Empty dataset for {symbol}")

        original_len = len(df)

        df = df.dropna(subset=["Open", "High", "Low", "Close"], how="all")
        df = df.ffill(limit=3)
        df = df.dropna(subset=["Close"])

        df = df.sort_index()
        df = df[~df.index.duplicated(keep="last")]

        now = pd.Timestamp.utcnow()
        future_mask = df.index > now + pd.Timedelta(minutes=5)
        if future_mask.any():
            logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps")
            df = df[~future_mask]

        invalid_ohlc = (
            (df["High"] < df["Low"]) |
            (df["High"] < df["Open"]) |
            (df["High"] < df["Close"]) |
            (df["Low"] > df["Open"]) |
            (df["Low"] > df["Close"])
        )
        if invalid_ohlc.any():
            logger.warning(f"{symbol}: Fixing {invalid_ohlc.sum()} OHLC-inconsistent bars")
            for idx in df[invalid_ohlc].index:
                vals = [df.loc[idx, "Open"], df.loc[idx, "Close"]]
                df.loc[idx, "High"] = max(vals + [df.loc[idx, "High"]])
                df.loc[idx, "Low"]  = min(vals + [df.loc[idx, "Low"]])

        returns = df["Close"].pct_change()
        std_ret = returns.std()
        if std_ret > 0:
            outlier_mask = (returns - returns.mean()).abs() > 10 * std_ret
            if outlier_mask.any():
                logger.warning(f"{symbol}: Removing {outlier_mask.sum()} extreme outlier bars")
                df = df[~outlier_mask]

        min_required = max(
            self.config.technical.ema_trend,
            200,
        ) + 10

        if len(df) < min_required:
            logger.warning(
                f"{symbol} {timeframe.value}: Only {len(df)} bars "
                f"(need {min_required}). Some indicators will be partial."
            )

        cleaned = original_len - len(df)
        if cleaned > 0:
            logger.info(f"{symbol}: Cleaned {cleaned}/{original_len} bars")

        return df