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 threading
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] = {}
        # The bot shares one provider across a thread pool; cache eviction
        # and the throttle deque are check-then-act and need locking.
        self._cache_lock = threading.Lock()
        self._throttle_lock = threading.Lock()
        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)
        with self._cache_lock:
            cached = self._cache.get(cache_key)
            cached_at = self._cache_timestamps.get(cache_key)
        if (
            cached is not None
            and cached_at is not None
            and (now - cached_at) < cache_ttl
            and not cached.empty
        ):
            if (self._is_data_fresh(cached, symbol, timeframe)
                    and self._has_latest_closed_bar(cached, timeframe)):
                return cached
            logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching")
            with self._cache_lock:
                self._cache.pop(cache_key, None)
                self._cache_timestamps.pop(cache_key, None)

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

        with self._cache_lock:
            self._cache[cache_key] = df
            self._cache_timestamps[cache_key] = now
        return df

    def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe) -> bool:
        """Cached data must include the most recent fully closed bar.

        The TTL alone is wall-clock based: a snapshot fetched mid-candle is
        still within TTL at the next candle boundary but is missing the bar
        that just closed, so every boundary scan would analyze — and run the
        bar-close session check against — the previous bar.
        """
        now = pd.Timestamp.now(tz="UTC")
        if _is_forex_weekend(now):
            return True
        candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
        dur = pd.Timedelta(seconds=candle_seconds)
        expected_label = now.floor(dur) - dur
        return df.index[-1] >= expected_label

    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:
            with self._throttle_lock:
                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

                used = len(self._request_times)
                wait = (self._request_times[0] + self.RATE_WINDOW) - now + 0.2
            logger.info(
                f"Rate limit: {used}/{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:
            dev = returns - returns.mean()
            outlier_mask = dev.abs() > 10 * std_ret
            # A spike bar makes the NEXT bar's return look extreme too (it is
            # measured against the spike). If the previous bar is flagged and
            # this bar reverses direction, it is the legitimate snap-back —
            # keep it, or every spike would punch a two-bar hole in the data.
            recovery = (
                outlier_mask
                & outlier_mask.shift(1, fill_value=False)
                & (dev * dev.shift(1) < 0)
            )
            outlier_mask = outlier_mask & ~recovery
            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