""" ForexBot2 — Twelve Data provider ================================= OHLCV via the official `twelvedata` SDK plus streaming quotes over the Twelve Data WebSocket. All shared behavior (cache, validation, boundary guarantees, range pagination) lives in `BaseDataProvider`. """ import asyncio import json import logging import threading import time from collections import deque from datetime import datetime from typing import AsyncIterator, Callable, Optional, Sequence, Tuple import pandas as pd from twelvedata import TDClient from app_config import provider_api_key from config import PAIR_INFO, PlatformConfig, Timeframe from providers.base import BaseDataProvider, DataIntegrityError logger = logging.getLogger(__name__) class TwelveDataProvider(BaseDataProvider): """Twelve Data SDK client with credit throttling and WS quotes.""" name = "twelve_data" 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", } RATE_LIMIT = 55 RATE_WINDOW = 60.0 WS_URL = "wss://ws.twelvedata.com/v1/quotes/price?apikey={api_key}" def __init__(self, config: PlatformConfig): super().__init__(config) self._api_key = provider_api_key(provider="twelve_data") or None 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:]}" # ── Raw fetch ────────────────────────────────────────────────── def _fetch_raw( self, symbol: str, timeframe: Timeframe, outputsize: int, start: Optional[datetime] = None, end: Optional[datetime] = None, ) -> pd.DataFrame: return self._fetch_twelve_data( symbol, timeframe, outputsize, start_date=start.strftime("%Y-%m-%d %H:%M:%S") if start else None, end_date=end.strftime("%Y-%m-%d %H:%M:%S") if end else None, ) 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, self.MAX_BARS_PER_REQUEST), "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 # ── Streaming quotes ─────────────────────────────────────────── supports_streaming = True @staticmethod def _parse_ws_price(raw: str) -> Optional[Tuple[str, float]]: """Parse one WS message; return (ws_symbol, price) for price events, None for everything else (logged where noteworthy).""" try: msg = json.loads(raw) except json.JSONDecodeError: logger.warning("WebSocket: malformed JSON, skipping") return None event = msg.get("event") if event == "subscribe-status": fails = msg.get("fails") or [] if fails: logger.warning(f"WebSocket: failed to subscribe: {fails}") return None if event != "price": return None ws_symbol = msg.get("symbol") try: price = float(msg["price"]) except (KeyError, ValueError, TypeError) as e: logger.warning(f"WebSocket: bad price data for {ws_symbol}: {e}") return None return ws_symbol, price @staticmethod async def _ws_heartbeat(ws) -> None: """Twelve Data drops idle WS connections; heartbeat every 10s keeps it alive through quiet periods (weekends, low-liquidity hours).""" while True: await asyncio.sleep(10) await ws.send(json.dumps({"action": "heartbeat"})) async def iter_ticks( self, symbols: Sequence[str], on_connect: Optional[Callable[[], None]] = None, ) -> AsyncIterator[Tuple[str, float]]: """One connection's lifetime of (canonical_symbol, price) ticks. The TD WS speaks display symbols ("XAU/USD"); ticks are mapped back to the canonical pair symbols passed in. The caller owns the reconnect/backoff loop around this generator. """ import websockets ws_to_pair = {PAIR_INFO[s]["display"]: s for s in symbols} url = self.WS_URL.format(api_key=self._api_key or "") async with websockets.connect(url) as ws: subscribe = { "action": "subscribe", "params": {"symbols": ",".join(ws_to_pair)}, } await ws.send(json.dumps(subscribe)) logger.info(f"WebSocket connected, subscribed to {len(ws_to_pair)} symbols") if on_connect is not None: on_connect() hb_task = asyncio.create_task(self._ws_heartbeat(ws)) try: async for raw in ws: parsed = self._parse_ws_price(raw) if parsed is None: continue pair_symbol = ws_to_pair.get(parsed[0]) if pair_symbol is None: continue yield pair_symbol, parsed[1] finally: hb_task.cancel()