Harwinder Sekhon
44cadf7a9f
Resolves the Twelve Data API key through app_config.provider_api_key() (env → config.json → no default) instead of a hardcoded field on PlatformConfig. Adds data_provider.PROVIDER_REGISTRY + make_data_provider() so signal_aggregator and the backtester build the active provider through one seam; telegram_bot.py gets the same one-line WS-key fix now (per Task 4's already-planned edit for that line) so the suite doesn't crash on import between this commit and Task 4's full wiring pass. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VPNF37MP9Kuv62B1VBrZ1b
401 lines
14 KiB
Python
Executable File
401 lines
14 KiB
Python
Executable File
"""
|
|
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_latest_price(self, symbol: str = XAUUSD) -> Optional[float]:
|
|
try:
|
|
df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1)
|
|
if df is not None and not df.empty:
|
|
return float(df["Close"].iloc[-1])
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
def get_spread_estimate(self, symbol: str = XAUUSD) -> float:
|
|
try:
|
|
df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1)
|
|
if df is not None and not df.empty:
|
|
return float((df["High"] - df["Low"]).median())
|
|
except Exception:
|
|
pass
|
|
info = PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])
|
|
return info["pip_value"]
|
|
|
|
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)
|