ForexBot2/providers/oanda.py

"""
ForexBot2 — Oanda v20 data provider
====================================
OHLCV candles from the Oanda v20 REST API. All shared behavior (cache,
validation, boundary guarantees, range pagination) lives in
`BaseDataProvider`; this module only maps symbols/granularities, parses
Oanda's candle JSON, and exposes the real bid/ask spread from MBA candles.
"""

import logging
import time
from datetime import datetime
from typing import Dict, Optional

import pandas as pd
import requests

from app_config import oanda_api_token, oanda_base_url
from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD

from providers.base import BaseDataProvider, DataIntegrityError

logger = logging.getLogger(__name__)


class OandaDataError(DataIntegrityError):
    """Raised when an Oanda data fetch fails or returns invalid data."""


class OandaDataProvider(BaseDataProvider):
    """Oanda v20 REST candle provider."""

    name = "oanda"
    DATA_ERROR = OandaDataError
    # Transient-fault retry ladder (see _fetch_oanda_candles): sleeps
    # between attempts, plus a wall-clock cap on the whole sequence.
    RETRY_DELAYS_S = (2.0, 5.0, 10.0)
    RETRY_BUDGET_S = 45.0
    RETRY_DELAY_S = RETRY_DELAYS_S[0]  # first delay; kept for callers/tests

    OANDA_GRANULARITY = {
        Timeframe.M1: "M1",
        Timeframe.M5: "M5",
        Timeframe.M15: "M15",
        Timeframe.M30: "M30",
        Timeframe.H1: "H1",
        Timeframe.H4: "H4",
        Timeframe.D1: "D",
        Timeframe.W1: "W",
    }

    OANDA_INSTRUMENT = {
        "XAU/USD": "XAU_USD",
    }

    def __init__(
        self,
        config: PlatformConfig,
        api_token: Optional[str] = None,
        base_url: Optional[str] = None,
    ):
        super().__init__(config)
        # Credentials default to the config.json `oanda` block / env vars
        # so the provider registry can build this class with config only.
        if api_token is None:
            api_token = oanda_api_token()
        if base_url is None:
            base_url = oanda_base_url()
        self._base_url = base_url.rstrip("/")
        self._session = requests.Session()
        self._session.headers.update({
            "Authorization": f"Bearer {api_token}",
            "Accept-Datetime-Format": "RFC3339",
            "Content-Type": "application/json",
        })
        self._spread_cache: Dict[str, float] = {}

    # ── Public extras ──────────────────────────────────────────────

    def get_spread(self, symbol: str = XAUUSD) -> float:
        return self._spread_cache.get(
            symbol, PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])["default_spread"])

    # ── Raw fetch ──────────────────────────────────────────────────

    def _fetch_raw(
        self,
        symbol: str,
        timeframe: Timeframe,
        outputsize: int,
        start: Optional[datetime] = None,
        end: Optional[datetime] = None,
    ) -> pd.DataFrame:
        if start is not None or end is not None:
            return self._fetch_oanda_candles(
                symbol, timeframe, outputsize, start, end)
        return self._fetch_oanda_candles(symbol, timeframe, outputsize)

    def _to_oanda_instrument(self, symbol: str) -> str:
        try:
            return self.OANDA_INSTRUMENT[symbol]
        except KeyError:
            raise OandaDataError(
                f"Unsupported symbol {symbol!r} for Oanda") from None

    def _fetch_oanda_candles(
        self,
        symbol: str,
        timeframe: Timeframe,
        count: int,
        start: Optional[datetime] = None,
        end: Optional[datetime] = None,
    ) -> pd.DataFrame:
        instrument = self._to_oanda_instrument(symbol)
        granularity = self.OANDA_GRANULARITY.get(timeframe)
        if granularity is None:
            raise OandaDataError(
                f"Unsupported timeframe {timeframe} for Oanda")

        url = f"{self._base_url}/v3/instruments/{instrument}/candles"
        params = {
            "granularity": granularity,
            "price": "MBA",
            "dailyAlignment": 0,
            "alignmentTimezone": "Etc/UTC",
        }
        # v20 rejects `count` combined with both `from` and `to`; a date
        # window replaces it (used by fetch_range pagination).
        if start is not None and end is not None:
            params["from"] = pd.Timestamp(start).isoformat()
            params["to"] = pd.Timestamp(end).isoformat()
        else:
            params["count"] = min(count, self.MAX_BARS_PER_REQUEST)

        # Retry ladder for transient faults: Oanda's sporadic spurious
        # 401s (~15-20/day on the practice API — the same token succeeds
        # on adjacent requests), 5xx and network errors. Other 4xx (bad
        # request, 429) are never retried; a real credential problem
        # fails every attempt and raises as before. The 2026-09-07
        # signal-log audit found 11 boundary scans in two days that lost
        # their primary frame at xx:00:22-:26 UTC: a single 2 s retry
        # could not outlast a 10-15 s provider blip at the top of the
        # hour, and a scan without its primary frame logs a price-0
        # NEUTRAL row. RETRY_BUDGET_S bounds the whole sequence so a scan
        # (three fetches) still finishes well inside an M5 bar.
        last_err: Optional[OandaDataError] = None
        resp = None
        attempts = 1 + len(self.RETRY_DELAYS_S)
        started = time.monotonic()
        for attempt in range(attempts):
            if attempt:
                delay = self.RETRY_DELAYS_S[attempt - 1]
                if time.monotonic() - started + delay > self.RETRY_BUDGET_S:
                    logger.warning(
                        f"Oanda fetch for {instrument}: retry budget "
                        f"({self.RETRY_BUDGET_S:.0f}s) exhausted after "
                        f"{attempt} attempt(s) — giving up")
                    break
                time.sleep(delay)
            try:
                resp = self._session.get(url, params=params, timeout=30)
            except requests.RequestException as e:
                last_err = OandaDataError(
                    f"Oanda candle fetch failed for {instrument}: {e}")
                last_err.__cause__ = e
                resp = None
                logger.warning(
                    f"Oanda fetch for {instrument} attempt "
                    f"{attempt + 1}/{attempts} failed: {e}")
                continue
            if resp.status_code == 401 or resp.status_code >= 500:
                last_err = OandaDataError(
                    f"Oanda candle fetch failed for {instrument}: "
                    f"HTTP {resp.status_code}: {resp.text[:300]}")
                logger.warning(
                    f"Oanda fetch for {instrument} attempt "
                    f"{attempt + 1}/{attempts} got HTTP {resp.status_code}")
                resp = None
                continue
            break
        if resp is None:
            raise last_err

        if resp.status_code != 200:
            raise OandaDataError(
                f"Oanda candle fetch failed for {instrument}: "
                f"HTTP {resp.status_code}: {resp.text[:300]}")

        try:
            data = resp.json()
        except ValueError as e:
            raise OandaDataError(
                f"Oanda candle response for {instrument} is not valid "
                f"JSON: {e}") from e
        candles = data.get("candles", [])
        if not candles:
            raise OandaDataError(
                f"Empty candle response from Oanda for {instrument}")

        rows = []
        last_bid_close = None
        last_ask_close = None
        skipped_malformed = 0
        for c in candles:
            if not c.get("complete", False):
                continue
            mid = c.get("mid", {})
            bid = c.get("bid", {})
            ask = c.get("ask", {})
            # A candle missing any OHLC field must be SKIPPED, never
            # defaulted to 0.0: a zero price flows through cleaning as a
            # "legit" bar and poisons ATR/Keltner/SL math — and a
            # mid-frame zero Close makes pct_change() emit inf, which
            # NaNs the std and disarms the 10σ outlier filter for the
            # whole frame (2026-08-04 audit, I45).
            if any(mid.get(k) is None for k in ("o", "h", "l", "c")):
                skipped_malformed += 1
                continue
            rows.append({
                "Datetime": pd.Timestamp(c["time"]).tz_convert("UTC"),
                "Open": float(mid["o"]),
                "High": float(mid["h"]),
                "Low": float(mid["l"]),
                "Close": float(mid["c"]),
                "Volume": int(c.get("volume", 0)),
            })
            if bid and ask:
                last_bid_close = float(bid.get("c", 0))
                last_ask_close = float(ask.get("c", 0))
        if skipped_malformed:
            logger.warning(
                "Oanda returned %d malformed candle(s) for %s (missing "
                "OHLC fields) — skipped", skipped_malformed, instrument)

        if not rows:
            raise OandaDataError(
                f"No complete candles from Oanda for {instrument}")

        if last_bid_close and last_ask_close:
            self._spread_cache[symbol] = round(
                last_ask_close - last_bid_close, 4)

        df = pd.DataFrame(rows)
        df = df.set_index("Datetime")
        df = df.sort_index()
        if df.index.tz is None:
            df.index = df.index.tz_localize("UTC")
        df.index.name = "Datetime"

        for col in ("Open", "High", "Low", "Close"):
            df[col] = pd.to_numeric(df[col], errors="coerce")
        df["Volume"] = pd.to_numeric(df["Volume"], errors="coerce").fillna(0).astype(float)

        logger.debug(
            "Fetched %d bars for %s @ %s from Oanda",
            len(df), instrument, granularity)
        return df