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, USDJPY, PAIR_INFO

logger = logging.getLogger(__name__)


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 = USDJPY,
        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:
                return cached

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

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

    def get_latest_price(self, symbol: str = USDJPY) -> 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 = USDJPY) -> 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[USDJPY])
        return info["pip_value"]

    def get_pair_info(self, symbol: str = USDJPY) -> Dict:
        info = PAIR_INFO.get(symbol, PAIR_INFO[USDJPY])
        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"])

        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,
            self.config.technical.ichimoku_senkou_b,
            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."
            )

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

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

        return df