harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Provider abstraction: providers/ package with BaseDataProvider template

Browse Source 
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
This commit is contained in:
Harwinder Sekhon
2026-07-28 20:09:00 -04:00
parent f760e9070a
commit 083a4e8d63
17 changed files with 1573 additions and 935 deletions
Download Patch File Download Diff File Expand all files Collapse all files
data_provider.py
+20 -399
View File
@@ -1,403 +1,24 @@
"""
ForexBot2 — Data Provider (Twelve Data SDK)
==========================================================
Fetches and validates OHLCV via the official `twelvedata` Python SDK.
ForexBot2 — Data Provider (back-compat shim)
=============================================
The provider implementations live in the `providers/` package:
providers/base.py BaseDataProvider (shared contract + behavior)
providers/twelve_data.py TwelveDataProvider (SDK fetch, throttle, WS)
providers/oanda.py OandaDataProvider (v20 REST, real spread)
`ForexDataProvider` is the historical name of the Twelve Data provider and
is kept as an alias so existing imports keep working.
"""
import logging
import threading
import time
from collections import deque
from typing import Dict, Optional
from providers import ( # noqa: F401
BaseDataProvider,
DataIntegrityError,
PROVIDER_REGISTRY,
TwelveDataProvider,
make_data_provider,
_is_forex_weekend,
)
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
# Grace wait before the one retry when a fresh response at a candle
# boundary is missing the just-closed bar (feed latency).
FRESH_FETCH_RETRY_SECONDS = 3.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)
# The closed-bar guarantee must hold for FRESH responses too, not
# just cache hits: the scanner fires seconds after the boundary,
# and a lagging feed can answer without the just-closed bar — the
# scan would silently analyze (and session-check) the previous
# bar. Give the feed one short grace retry, then proceed loudly.
if (not df.empty and not self._has_latest_closed_bar(df, timeframe)):
time.sleep(self.FRESH_FETCH_RETRY_SECONDS)
retry = self._fetch_twelve_data(symbol, timeframe, outputsize)
retry = self._drop_forming_candle(retry, timeframe)
retry = self._validate_and_clean(retry, symbol, timeframe)
if retry is not None and not retry.empty:
df = retry
if not self._has_latest_closed_bar(df, timeframe):
logger.warning(
f"{symbol} {timeframe.value}: feed lagging — just-closed bar "
f"still missing after retry; this scan may run one bar behind"
)
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."""
with self._cache_lock:
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)
# Historical name for the Twelve Data provider.
ForexDataProvider = TwelveDataProvider