"""
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
from zoneinfo import ZoneInfo

import pandas as pd

from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD

logger = logging.getLogger(__name__)

# The gold/FX market week is defined in New York time (see
# _is_forex_weekend); a fixed UTC hour is wrong for half the year.
_NY_TZ = ZoneInfo("America/New_York")


def _is_forex_weekend(now=None) -> bool:
    """Gold/FX market closed for the weekend.

    The market week is anchored to NEW YORK time — Friday 17:00 ET →
    Sunday 18:00 ET — not a fixed UTC hour. During US DST the feed
    actually stops at 21:00 UTC (measured on 5 years of Oanda bars:
    159/162 summer Fridays; winter close 22:00, winter reopen Sunday
    23:00 UTC). The old hardcoded Fri 22:00→Sun 22:00 UTC rule produced
    two false-"STALE DATA/feed lagging" hours every week (2026-08-04
    audit I43, fixed per owner decision).

    Kept as three deliberate copies (here, signal_aggregator.py,
    gpu_backtest._is_xau_blocked_by_calendar) — a Module 21 test asserts
    they agree hour-by-hour across DST transitions.
    """
    if now is None:
        now = pd.Timestamp.now(tz="UTC")
    ny = now.astimezone(_NY_TZ)
    wd = ny.weekday()
    if wd == 4:                      # Friday
        return ny.hour >= 17
    if wd == 5:                      # Saturday
        return True
    if wd == 6:                      # Sunday
        return ny.hour < 18
    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

    # Signal-side bad-print filter (see _validate_and_clean): a bar is a
    # candidate when its close-to-close return sits more than
    # OUTLIER_SIGMA standard deviations from the frame mean, and is
    # dropped only when the NEXT close snaps back at least
    # OUTLIER_REVERT_FRACTION of the way to the pre-spike close.
    OUTLIER_SIGMA = 10.0
    OUTLIER_REVERT_FRACTION = 0.8

    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:
            # A 10σ close-to-close move is NOT bad data by default. On
            # three years of Oanda XAU bars every such bar was a real
            # news/reopen move (NFP, CPI, Sunday gap — 2026-09-07 audit:
            # 155 flagged M5 bars, 7 snapped back), and dropping them hid
            # the most violent bar of the day from ATR/exhaustion math —
            # while the backtester (whole-frame σ) dropped a DIFFERENT
            # set than live (fetch-window σ). A bad print has the one
            # signature a real move lacks: the NEXT close returns to the
            # pre-spike level. Drop a candidate only on that observed
            # snap-back; the newest bar (no successor yet) is kept as
            # real, and the snap-back bar itself is never a candidate
            # (its residual against the pre-spike close is small).
            dev = returns - returns.mean()
            candidate = dev.abs() > self.OUTLIER_SIGMA * std_ret
            close = df["Close"]
            spike = close - close.shift(1)
            residual = close.shift(-1) - close.shift(1)
            snapped_back = residual.notna() & (
                residual.abs()
                <= (1.0 - self.OUTLIER_REVERT_FRACTION) * spike.abs()
            )
            outlier_mask = candidate & snapped_back
            if outlier_mask.any():
                logger.warning(
                    f"{symbol}: Removing {int(outlier_mask.sum())} bad-print "
                    f"bar(s) (>{self.OUTLIER_SIGMA:.0f}σ move that snapped back)")
                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