Harwinder Sekhon
083a4e8d63
One class per provider (TwelveData, Oanda, Tiingo), all subclassing BaseDataProvider which owns cache, validation/cleaning, freshness + boundary guarantees, fetch_range pagination, and streaming stubs — subclasses implement only _fetch_raw. config.json data_providers.active selects the live feed; data_provider.py / oanda_provider.py stay as back-compat shims. Module 16 provider-abstraction tests (253 total green). Claude-Session: https://claude.ai/code/session_01Mc7XKYyPySVXiUpcN82QEK
239 lines
8.2 KiB
Python
239 lines
8.2 KiB
Python
"""
|
|
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()
|