"""
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] = {}
        # The bot shares one provider across a thread pool; cache eviction
        # is check-then-act and needs locking.
        self._cache_lock = threading.Lock()

    # ── 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}"
        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,
                self.MAX_BARS_PER_REQUEST),
        )
        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
        return df

    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
        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:
                # Providers raise for windows with no data (weekends,
                # holidays, asset gaps). Skip the window, step back, retry
                # — but bail after a few consecutive empties.
                logger.warning(
                    f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}"
                )
                empty_streak += 1
                if empty_streak >= 3:
                    break
                cursor_end = cursor_start - timedelta(seconds=1)
                continue

            if chunk is None or chunk.empty:
                empty_streak += 1
                if empty_streak >= 3:
                    break
                cursor_end = cursor_start - timedelta(seconds=1)
                continue

            empty_streak = 0
            chunks.append(chunk)
            # Step back; dedup on concat.
            cursor_end = chunk.index[0] - timedelta(seconds=1)
            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)
        return df[(df.index >= start_ts) & (df.index <= end_ts)]

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

        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