ForexBot2/data_provider.py

"""
ForexBot2 — Data Provider (Twelve Data, USD/JPY only)
=======================================================
Fetches and validates OHLCV from Twelve Data. No synthetic fallback —
if the API fails the caller learns about it. Adds a parquet on-disk
cache so the backtest worker pool reuses one download instead of
hammering Twelve Data 27 times during a parameter sweep.
"""

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

import pandas as pd
import requests

from config import PlatformConfig, Timeframe, USDJPY, DISPLAY, PIP_VALUE

logger = logging.getLogger(__name__)


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


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

    TD_BASE_URL = "https://api.twelvedata.com"

    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,
    }

    # Free plan: 8 credits / minute
    RATE_LIMIT = 8
    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._session = requests.Session()
        self._session.headers.update({
            "User-Agent": "Mozilla/5.0 (compatible; ForexIntel/1.0)"
        })

    @staticmethod
    def _to_td_symbol(symbol: str) -> str:
        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()

        if (
            cache_key in self._cache_timestamps
            and (now - self._cache_timestamps[cache_key]) < self.config.refresh_interval
            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 fetch_full_history(
        self,
        symbol: str = USDJPY,
        timeframe: Timeframe = Timeframe.H1,
        max_bars: int = 5000,
        use_disk_cache: bool = True,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
    ) -> pd.DataFrame:
        """
        Pull the largest window Twelve Data allows in one call.
        On-disk parquet cache prevents re-download across worker subprocesses
        during an optimizer sweep.
        """
        cache_path = self._disk_cache_path(symbol, timeframe, start_date, end_date)

        if use_disk_cache and os.path.exists(cache_path):
            try:
                df = pd.read_parquet(cache_path)
                if df.index.tz is None:
                    df.index = df.index.tz_localize("UTC")
                logger.info(
                    f"Loaded {len(df)} bars from disk cache: {cache_path}"
                )
                return df
            except Exception as e:
                logger.warning(f"Disk cache read failed ({e}); refetching.")

        outputsize = min(max_bars, 5000)
        df = self._fetch_twelve_data(
            symbol, timeframe, outputsize,
            start_date=start_date, end_date=end_date,
        )
        df = self._validate_and_clean(df, symbol, timeframe)

        if use_disk_cache:
            os.makedirs(os.path.dirname(cache_path), exist_ok=True)
            try:
                df.to_parquet(cache_path)
                logger.info(f"Cached {len(df)} bars → {cache_path}")
            except Exception as e:
                logger.warning(f"Disk cache write failed: {e}")

        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:
        """Estimate spread from recent M5 H-L median; fallback to 1 pip on JPY."""
        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
        # Fallback: 1 pip = 0.01 for JPY pairs
        return PIP_VALUE

    def get_pair_info(self, symbol: str = USDJPY) -> Dict:
        return {
            "symbol": symbol,
            "display": DISPLAY,
            "pip_value": PIP_VALUE,
            "is_jpy_pair": True,
        }

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

    def _disk_cache_path(
        self,
        symbol: str,
        timeframe: Timeframe,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
    ) -> str:
        clean = symbol.replace("=X", "").lower()
        suffix = ""
        if start_date or end_date:
            suffix = f"_{start_date or 'min'}_{end_date or 'max'}"
        return os.path.join(
            self.config.backtest.cache_dir,
            f"{clean}_{timeframe.value}{suffix}.parquet",
        )

    def _throttle(self, credits: int = 1) -> None:
        """Free tier: 8 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 not self._api_key:
            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)

        params = {
            "symbol": td_symbol,
            "interval": interval,
            "outputsize": outputsize,
            "apikey": self._api_key,
            "timezone": "UTC",
            "format": "json",
        }
        if start_date:
            params["start_date"] = start_date
        if end_date:
            params["end_date"] = end_date

        resp = self._session.get(
            f"{self.TD_BASE_URL}/time_series",
            params=params,
            timeout=30,
        )
        resp.raise_for_status()
        data = resp.json()

        if data.get("status") == "error":
            raise DataIntegrityError(
                f"Twelve Data error for {td_symbol}: {data.get('message', 'unknown')}"
            )

        values = data.get("values")
        if not values:
            raise DataIntegrityError(f"Empty response from Twelve Data for {td_symbol}")

        records = [
            {
                "Datetime": pd.Timestamp(v["datetime"], tz="UTC"),
                "Open":   float(v["open"]),
                "High":   float(v["high"]),
                "Low":    float(v["low"]),
                "Close":  float(v["close"]),
                "Volume": float(v.get("volume") or 0),
            }
            for v in reversed(values)
        ]

        df = pd.DataFrame(records).set_index("Datetime")
        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