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


def _is_daily_maintenance(now=None) -> bool:
    """Gold's daily settlement pause: Monday-Thursday 17:00-17:59 New York
    (21:00 UTC in US summer, 22:00 in winter). No bar carries a label
    inside this hour. Friday 17:00 is the weekend (see _is_forex_weekend).
    Same rule as advisory_outcome._has_continuity's daily_maintenance."""
    if now is None:
        now = pd.Timestamp.now(tz="UTC")
    ny = now.astimezone(_NY_TZ)
    return ny.weekday() < 4 and ny.hour == 17


def _is_scheduled_closure(now=None) -> bool:
    """The feed is legitimately silent: weekend or daily settlement pause."""
    return _is_forex_weekend(now) or _is_daily_maintenance(now)


def _scheduled_closed_seconds(start, end) -> float:
    """Seconds of the [start, end) span that fall inside a scheduled
    closure (weekend or the daily settlement pause).

    Bar age is measured from a bar LABEL, so at 21:30 UTC in summer the
    20:55 M5 bar looks 35 minutes old even though the feed could not have
    printed anything since 21:00. Subtracting the closed span keeps the
    3-candle freshness limit meaningful instead of failing it — and
    logging a false "feed lagging"/"STALE DATA" — every settlement hour
    (2026-09-13 audit: 61 suppressed scans on three nights).
    """
    hour = pd.Timedelta(hours=1)
    closed = 0.0
    t = start.floor("h")
    steps = 0
    while t < end and steps < 24 * 10:
        if _is_scheduled_closure(t):
            closed += max(
                0.0, (min(t + hour, end) - max(t, start)).total_seconds())
        t += hour
        steps += 1
    return closed


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

    #: Minimum bars per LIVE fetch on every timeframe (2026-09-10,
    #: indicator-seeding parity with the backtester). The engine's EMAs
    #: use ewm(adjust=False), so the first bar of the fetched frame keeps
    #: weight (1-2/201)^n in the 200-bar trend EMA. With the 60-day
    #: lookback, H4 fetched 380 bars (seed weight 2.3%) and D1 300 (5%)
    #: while the backtester seeds from its whole cached history. Measured
    #: on the deepest Oanda caches (267 H4 / 313 D1 bars sampled over two
    #: years, raw and cleaned frames alike): ema_trend flipped on 2/267 H4
    #: and 1/313 D1 bars at those depths (composite off by up to 9.5
    #: points) and on none from 600 bars up; 1000 leaves a seed weight of
    #: e^-10. M5/M15 (5000) and H1 (1460) already exceed it and are
    #: unchanged. One request either way (cap 5000).
    MIN_LIVE_BARS = 1000

    # 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 _live_outputsize(self, timeframe: Timeframe, lookback_days: int) -> int:
        """Bars requested for one live fetch: the lookback in bars, floored
        at MIN_LIVE_BARS (indicator seeding) and capped at the request cap."""
        wanted = int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20
        return max(self.MIN_LIVE_BARS, min(wanted, self.MAX_BARS_PER_REQUEST))

    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 = self._live_outputsize(timeframe, lookback_days)

        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
        window_errors = 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". RETRY THE SAME
                # WINDOW: until 2026-09-13 a failure advanced the cursor
                # past the window, so one or two NON-consecutive page
                # failures (Oanda's sporadic 401/5xx, even after the
                # provider's own retry ladder) silently dropped up to
                # days_per_chunk of bars — 15 days of M5 — out of the
                # middle of the frame, and only the HEAD was
                # coverage-checked below, so the backtester cached the hole
                # as complete history forever. Three failures on the SAME
                # window means it is unreachable: raise instead of
                # returning a frame with a hole in it.
                logger.warning(
                    f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}"
                )
                window_errors += 1
                if window_errors >= 3:
                    raise self.DATA_ERROR(
                        f"{symbol} {timeframe.value}: {window_errors} "
                        f"consecutive failures fetching window "
                        f"{cursor_start.date()}–{cursor_end.date()} — "
                        f"refusing to return truncated history "
                        f"(last error: {e})") from e
                continue        # re-request the SAME window

            window_errors = 0
            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
            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")
        # INTERIOR-gap audit (2026-09-13): the head check above is blind to
        # a hole in the MIDDLE of the frame, which is exactly what a
        # dropped page used to leave behind. Missing labels inside a
        # scheduled closure (weekend, daily settlement pause) are normal,
        # so only unscheduled runs are reported — and a real market holiday
        # (Christmas, Good Friday, July 4) looks identical to a lost page,
        # so this is LOUD BUT NOT FATAL and the frame is returned unchanged.
        if len(df) > 400_000:
            logger.info(
                f"{symbol} {timeframe.value}: interior-gap audit skipped "
                f"({len(df)} rows)")
        elif len(df) > 1:
            step = pd.Timedelta(seconds=self.CANDLE_SECONDS[timeframe])
            labels = df.index
            # Missing labels between each consecutive pair; >3 is the floor
            # worth a look (weekends/settlement hours leave 1-2 labels that
            # the NY-anchored closure rule does not cover).
            spans = ((labels[1:] - labels[:-1]) / step).astype("int64") - 1
            gaps = []
            for i in (spans > 3).nonzero()[0]:
                missing = sum(
                    1 for k in range(1, int(spans[i]) + 1)
                    if not _is_scheduled_closure(labels[i] + k * step))
                if missing > 3:
                    gaps.append((labels[i], labels[i + 1], missing))
            if gaps:
                logger.error(
                    f"{symbol} {timeframe.value}: {len(gaps)} interior "
                    f"gap(s) of >3 unscheduled bars in fetched history — "
                    f"first: {gaps[0]}; a page fetch may have been dropped "
                    f"or the market was closed (holiday); verify before "
                    f"trusting this window")
        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,
                               now=None) -> 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
        if now is None:
            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
        # No bar ever carries a label inside a scheduled closure, so the
        # newest bar that MUST be present is the last label before it.
        # Without this walk-back every scan from 21:05 to 22:00 UTC in
        # summer evicted the cache, refetched twice and logged a false
        # "feed lagging" (2026-09-13 audit).
        # A label counts as "inside" only when the closure covers the bar's
        # whole span: a D1 bar labelled Sunday 00:00 UTC starts in the
        # weekend but runs through the Sunday reopen, so it exists and
        # must be present (68 such bars in the 3-year Oanda D1 cache).
        steps = 0
        last_second = dur - pd.Timedelta(seconds=1)
        while (_is_scheduled_closure(expected_label)
               and _is_scheduled_closure(expected_label + last_second)
               and steps < 2000):
            expected_label -= dur
            steps += 1
        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,
                       now=None) -> bool:
        """Check if the last candle is within acceptable age for this timeframe."""
        if df is None or df.empty:
            return False

        if now is None:
            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
        # Scheduled closures (weekend, daily settlement pause) are dead
        # time in which no bar can print — they must not count as age.
        age_seconds = ((now - last_bar).total_seconds()
                       - _scheduled_closed_seconds(last_bar, now))

        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.now(tz="UTC")
        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