""" ForexBot2 — Data Provider (Twelve Data SDK) ========================================================== Fetches and validates OHLCV via the official `twelvedata` Python SDK. """ import logging import threading import time from collections import deque from typing import Dict, Optional import pandas as pd from twelvedata import TDClient from app_config import AppConfigError, get_app_config, provider_api_key from config import PlatformConfig, Timeframe, XAUUSD, PAIR_INFO 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 ForexDataProvider: """Twelve Data SDK client + on-disk parquet cache.""" TD_INTERVALS = { Timeframe.M1: "1min", Timeframe.M5: "5min", Timeframe.M15: "15min", Timeframe.M30: "30min", Timeframe.H1: "1h", Timeframe.H4: "4h", Timeframe.D1: "1day", Timeframe.W1: "1week", } 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, } RATE_LIMIT = 55 RATE_WINDOW = 60.0 def __init__(self, config: PlatformConfig): self.config = config self._api_key = provider_api_key() or None self._cache: Dict[str, pd.DataFrame] = {} self._cache_timestamps: Dict[str, float] = {} # The bot shares one provider across a thread pool; cache eviction # and the throttle deque are check-then-act and need locking. self._cache_lock = threading.Lock() self._throttle_lock = threading.Lock() self._request_times: deque = deque() self._td: Optional[TDClient] = ( TDClient(apikey=self._api_key) if self._api_key else None ) @staticmethod def _to_td_symbol(symbol: str) -> str: if "/" in symbol: return symbol clean = symbol.replace("=X", "") return f"{clean[:3]}/{clean[3:]}" # ── 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, 5000), ) df = self._fetch_twelve_data(symbol, timeframe, outputsize) df = self._drop_forming_candle(df, timeframe) df = self._validate_and_clean(df, symbol, timeframe) 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 _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. """ 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. Twelve Data's time_series includes 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. """ 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 clear_cache(self) -> int: """Wipe all cached data. Returns number of entries cleared.""" 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} # ── Private ──────────────────────────────────────────────────── def _throttle(self, credits: int = 1) -> None: """Pro tier: 55 credits / minute. Block until budget allows.""" while True: with self._throttle_lock: now = time.time() while self._request_times and now - self._request_times[0] >= self.RATE_WINDOW: self._request_times.popleft() available = self.RATE_LIMIT - len(self._request_times) if available >= credits: for _ in range(credits): self._request_times.append(time.time()) return used = len(self._request_times) wait = (self._request_times[0] + self.RATE_WINDOW) - now + 0.2 logger.info( f"Rate limit: {used}/{self.RATE_LIMIT} credits used " f"— waiting {wait:.1f}s..." ) time.sleep(wait) def _fetch_twelve_data( self, symbol: str, timeframe: Timeframe, outputsize: int, start_date: Optional[str] = None, end_date: Optional[str] = None, ) -> pd.DataFrame: if self._td is None: raise DataIntegrityError( "No data-provider API key. Set the TWELVE_DATA_API_KEY env " "var or data_providers.twelve_data.api_key in config.json." ) td_symbol = self._to_td_symbol(symbol) interval = self.TD_INTERVALS[timeframe] self._throttle(credits=1) kwargs = { "symbol": td_symbol, "interval": interval, "outputsize": min(outputsize, 5000), "timezone": "UTC", } if start_date: kwargs["start_date"] = start_date if end_date: kwargs["end_date"] = end_date try: df = self._td.time_series(**kwargs).as_pandas() except Exception as e: raise DataIntegrityError( f"Twelve Data fetch failed for {td_symbol}: {e}" ) from e if df is None or df.empty: raise DataIntegrityError(f"Empty response from Twelve Data for {td_symbol}") # SDK returns lowercase columns; engine expects capitalised. df = df.rename(columns={ "open": "Open", "high": "High", "low": "Low", "close": "Close", "volume": "Volume", }) for col in ("Open", "High", "Low", "Close", "Volume"): if col in df.columns: df[col] = pd.to_numeric(df[col], errors="coerce") if "Volume" not in df.columns: df["Volume"] = 0.0 # SDK returns newest-first; engine + indicators want oldest-first. df = df.sort_index() if df.index.tz is None: df.index = df.index.tz_localize("UTC") df.index.name = "Datetime" logger.debug(f"Fetched {len(df)} bars for {td_symbol} @ {interval}") return df def _validate_and_clean( self, df: pd.DataFrame, symbol: str, timeframe: Timeframe ) -> pd.DataFrame: if df is None or df.empty: raise DataIntegrityError(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 # ── Provider registry ───────────────────────────────────────────────── # config.json data_providers.active selects from here. Adding a provider # later = one adapter class + one entry + one config block. PROVIDER_REGISTRY: Dict[str, type] = {"twelve_data": ForexDataProvider} def make_data_provider(config: PlatformConfig) -> ForexDataProvider: """Build the ACTIVE data provider (config.json: data_providers.active).""" name = get_app_config().active_provider cls = PROVIDER_REGISTRY.get(name) if cls is None: raise AppConfigError( f'Unknown data provider "{name}"; ' f"registered: {sorted(PROVIDER_REGISTRY)}") return cls(config)