ForexBot2/providers/base.py

"""
ForexBot2 — Provider base class
================================
`BaseDataProvider` owns everything that must behave identically no matter
where the bars come from: in-memory caching with candle-duration TTL, the
closed-bar boundary guarantee, freshness checks, validation/cleaning, the
forming-candle drop, and paginated date-range fetching for the backtester.

A concrete provider implements only:
    _fetch_raw(symbol, timeframe, outputsize, start, end) -> pd.DataFrame
returning a normalized frame (capitalized OHLCV columns, UTC tz-aware
DatetimeIndex named "Datetime", oldest-first), plus optionally
`get_spread` and the streaming-quote capability (`supports_streaming` +
`iter_ticks`).

The signal pipeline (engine + aggregator + backtester) consumes ONLY the
public contract defined here, so swapping providers never touches signal
code.
"""

import logging
import threading
import time
from abc import ABC, abstractmethod
from datetime import datetime, timedelta
from typing import AsyncIterator, Callable, Dict, Optional, Sequence, Tuple

import pandas as pd

from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD

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 BaseDataProvider(ABC):
    """Shared OHLCV provider behavior; subclasses supply the raw fetch."""

    #: Registry name; overridden per provider.
    name: str = "base"

    #: Exception type this provider raises for fetch/validation failures.
    #: Provider-specific errors subclass DataIntegrityError so consumers
    #: can catch one type regardless of the active provider.
    DATA_ERROR = DataIntegrityError

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

    #: Hard per-request bar cap shared by Twelve Data and Oanda.
    MAX_BARS_PER_REQUEST = 5000

    # Grace wait before the one retry when a fresh response at a candle
    # boundary is missing the just-closed bar (feed latency).
    FRESH_FETCH_RETRY_SECONDS = 3.0

    def __init__(self, config: PlatformConfig):
        self.config = config
        self._cache: Dict[str, pd.DataFrame] = {}
        self._cache_timestamps: Dict[str, float] = {}
        # Requested fetch depth per key: a cached short frame must not
        # satisfy a caller asking for a longer lookback.
        self._cache_outputsize: Dict[str, int] = {}
        # The bot shares one provider across a thread pool; cache eviction
        # is check-then-act and needs locking.
        self._cache_lock = threading.Lock()
        # Single-flight: one API fetch per (symbol, TF) at a time; other
        # threads wait and then read the cache.
        self._inflight: Dict[str, threading.Event] = {}

    # ── Abstract: the only provider-specific piece ─────────────────

    @abstractmethod
    def _fetch_raw(
        self,
        symbol: str,
        timeframe: Timeframe,
        outputsize: int,
        start: Optional[datetime] = None,
        end: Optional[datetime] = None,
    ) -> pd.DataFrame:
        """Fetch up to `outputsize` bars, optionally bounded by start/end
        (tz-aware UTC datetimes). Must return a normalized frame:
        capitalized OHLCV columns, UTC DatetimeIndex named "Datetime",
        oldest-first. Raise self.DATA_ERROR on failure."""

    # ── 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}"
        outputsize = max(
            300,
            min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20,
                self.MAX_BARS_PER_REQUEST),
        )

        while True:
            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)
                cached_size = self._cache_outputsize.get(cache_key, 0)
            if (
                cached is not None
                and cached_at is not None
                and (now - cached_at) < cache_ttl
                and not cached.empty
                # A caller asking for MORE history than the cached fetch
                # requested must not be served the short frame silently.
                and cached_size >= outputsize
            ):
                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)
                    self._cache_outputsize.pop(cache_key, None)

            # Single-flight: concurrent scans for the same (symbol, TF)
            # must not each hit the API (the four TF loops fire together
            # at aligned boundaries and Oanda has no client throttle).
            # First caller claims the key and fetches; the rest wait,
            # then re-check the cache.
            with self._cache_lock:
                inflight = self._inflight.get(cache_key)
                if inflight is None:
                    self._inflight[cache_key] = threading.Event()
                    break
            inflight.wait(timeout=90)

        try:
            df = self._fetch_raw(symbol, timeframe, outputsize)
            df = self._drop_forming_candle(df, timeframe)
            df = self._validate_and_clean(df, symbol, timeframe)

            # The closed-bar guarantee must hold for FRESH responses too,
            # not just cache hits: the scanner fires seconds after the
            # boundary, and a lagging feed can answer without the
            # just-closed bar — the scan would silently analyze (and
            # session-check) the previous bar. Give the feed one short
            # grace retry, then proceed loudly.
            if not df.empty and not self._has_latest_closed_bar(df, timeframe):
                time.sleep(self.FRESH_FETCH_RETRY_SECONDS)
                try:
                    retry = self._fetch_raw(symbol, timeframe, outputsize)
                    retry = self._drop_forming_candle(retry, timeframe)
                    retry = self._validate_and_clean(retry, symbol, timeframe)
                    if retry is not None and not retry.empty:
                        df = retry
                except DataIntegrityError as e:
                    logger.warning(
                        f"{symbol} {timeframe.value}: boundary retry refetch "
                        f"failed ({e}) — keeping original data")
                if not self._has_latest_closed_bar(df, timeframe):
                    logger.warning(
                        f"{symbol} {timeframe.value}: feed lagging — just-closed bar "
                        f"still missing after retry; this scan may run one bar behind"
                    )

            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
                self._cache_outputsize[cache_key] = outputsize
            return df
        finally:
            # Release waiters even when the fetch raised — they re-check
            # the cache and (on a miss) claim the key themselves.
            with self._cache_lock:
                ev = self._inflight.pop(cache_key, None)
            if ev is not None:
                ev.set()

    def fetch_range(
        self,
        symbol: str,
        timeframe: Timeframe,
        start: datetime,
        end: datetime,
    ) -> pd.DataFrame:
        """Fetch the RAW closed-bar series between start/end, paginating
        under the per-request bar cap. Returns uncleaned bars (the
        backtester resolves trades on the raw price path); callers wanting
        the signal-side series run `validate_and_clean` on the result."""
        bars_per_day = self.BARS_PER_DAY[timeframe]
        days_per_chunk = max(1, int(4500 / bars_per_day))

        chunks = []
        cursor_end = end
        empty_streak = 0
        error_streak = 0
        while cursor_end > start:
            cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk))
            try:
                chunk = self._fetch_raw(
                    symbol, timeframe, self.MAX_BARS_PER_REQUEST,
                    start=cursor_start, end=cursor_end,
                )
            except Exception as e:
                # A raising chunk is a PROVIDER ERROR (auth, 429, 5xx,
                # malformed payload) — not "no data". Three in a row means
                # the rest of the history is unreachable: raise instead of
                # returning a silently truncated frame (the backtester
                # would treat it as complete).
                logger.warning(
                    f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}"
                )
                error_streak += 1
                if error_streak >= 3:
                    raise self.DATA_ERROR(
                        f"{symbol} {timeframe.value}: {error_streak} "
                        f"consecutive chunk failures walking back from "
                        f"{cursor_end.date()} — refusing to return "
                        f"truncated history (last error: {e})") from e
                cursor_end = cursor_start - timedelta(seconds=1)
                continue

            if chunk is None or chunk.empty:
                # Genuinely empty windows (weekends/holidays/asset gaps)
                # are normal; several in a row = start of history.
                empty_streak += 1
                if empty_streak >= 3:
                    break
                cursor_end = cursor_start - timedelta(seconds=1)
                continue

            empty_streak = 0
            error_streak = 0
            chunks.append(chunk)
            # Step back; dedup on concat.
            new_cursor = chunk.index[0] - timedelta(seconds=1)
            if new_cursor >= cursor_end:
                # A provider ignoring start/end would pin the cursor and
                # loop forever — refuse to spin.
                logger.warning(
                    f"{symbol} {timeframe.value}: pagination made no "
                    f"progress at {cursor_end} — stopping")
                break
            cursor_end = new_cursor
            if len(chunk) < 2:
                break

        if not chunks:
            raise RuntimeError(f"No data fetched for {symbol} {timeframe.value}")

        df = pd.concat(chunks).sort_index()
        df = df[~df.index.duplicated(keep="last")]
        # Fetches ending "now" include the still-forming candle — drop it
        # so callers only ever see closed bars.
        df = self._drop_forming_candle(df, timeframe)
        start_ts = pd.Timestamp(start)
        end_ts = pd.Timestamp(end)
        df = df[(df.index >= start_ts) & (df.index <= end_ts)]
        # Coverage sanity: a head materially later than the requested
        # start means history was truncated (empty-streak bail or provider
        # depth limit). Loud, not fatal — some assets genuinely start late.
        if not df.empty and (df.index[0] - start_ts) > pd.Timedelta(days=7):
            logger.error(
                f"{symbol} {timeframe.value}: fetched history starts "
                f"{df.index[0]} but {start_ts} was requested — "
                f"~{(df.index[0] - start_ts).days} days missing from the "
                f"head; treat downstream results accordingly")
        return df

    def is_data_fresh(
        self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
    ) -> bool:
        """Public freshness check (last bar within 3 candle durations)."""
        return self._is_data_fresh(df, symbol, timeframe)

    def validate_and_clean(
        self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
    ) -> pd.DataFrame:
        """Public entry to the shared cleaning pipeline."""
        return self._validate_and_clean(df, symbol, timeframe)

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

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

    def get_spread(self, symbol: str = XAUUSD) -> float:
        """Latest observed round-trip spread; providers without live
        bid/ask fall back to the pair's assumed default."""
        return PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])["default_spread"]

    # ── Streaming-quote capability (optional) ──────────────────────

    #: True when the provider can stream live ticks via `iter_ticks`.
    supports_streaming: bool = False

    def iter_ticks(
        self,
        symbols: Sequence[str],
        on_connect: Optional[Callable[[], None]] = None,
    ) -> AsyncIterator[Tuple[str, float]]:
        """Async generator yielding (canonical_symbol, price) ticks for one
        connection's lifetime; the caller owns reconnect/backoff. Providers
        that support it set `supports_streaming = True` and override."""
        raise NotImplementedError(
            f"{self.name} provider does not support streaming quotes")

    # ── Shared internals ───────────────────────────────────────────

    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.
        """
        if df is None or df.empty:
            return False
        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.

        Some feeds include 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. Feeds that only
        return complete candles pass through unchanged.
        """
        if df is None or df.empty:
            return df
        candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
        now = pd.Timestamp.now(tz="UTC")
        # Loop, not a single slice: a misbehaving feed can return MORE
        # than one not-yet-closed bar, and a single iloc[:-1] would let
        # the residual forming bar through cleaning (future-tolerance is
        # +5 min) straight into the signal engine.
        dropped = 0
        while (not df.empty
               and df.index[-1] + pd.Timedelta(seconds=candle_seconds) > now):
            df = df.iloc[:-1]
            dropped += 1
        if dropped > 1:
            logger.warning(
                "%s: feed returned %d forming/future bars — all dropped",
                timeframe.value, dropped)
        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 _validate_and_clean(
        self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
    ) -> pd.DataFrame:
        if df is None or df.empty:
            raise self.DATA_ERROR(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]

        # Non-positive prices are corrupt, never real gold quotes. Reject
        # the row BEFORE the consistency repair below — repairing High/Low
        # from a corrupt 0.0 Open/Close would propagate the zero into
        # previously-valid fields (2026-08-04 audit, I45).
        nonpositive = (df[["Open", "High", "Low", "Close"]] <= 0).any(axis=1)
        if nonpositive.any():
            logger.warning(
                f"{symbol}: Dropping {nonpositive.sum()} bars with "
                f"non-positive prices (corrupt feed data)")
            df = df[~nonpositive]
            if df.empty:
                raise self.DATA_ERROR(
                    f"{symbol}: all bars had non-positive prices")

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

        # inf returns (division by a zero/na Close) would NaN the std and
        # silently disarm the whole 10σ pass — neutralize them first.
        returns = df["Close"].pct_change()
        returns = returns.replace([float("inf"), float("-inf")], float("nan"))
        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