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
424 lines
16 KiB
Python
424 lines
16 KiB
Python
"""
|
||
ForexBot2 — Provider base class
|
||
================================
|
||
`BaseDataProvider` owns everything that must behave identically no matter
|
||
where the bars come from: in-memory caching with candle-duration TTL, the
|
||
closed-bar boundary guarantee, freshness checks, validation/cleaning, the
|
||
forming-candle drop, and paginated date-range fetching for the backtester.
|
||
|
||
A concrete provider implements only:
|
||
_fetch_raw(symbol, timeframe, outputsize, start, end) -> pd.DataFrame
|
||
returning a normalized frame (capitalized OHLCV columns, UTC tz-aware
|
||
DatetimeIndex named "Datetime", oldest-first), plus optionally
|
||
`get_spread` and the streaming-quote capability (`supports_streaming` +
|
||
`iter_ticks`).
|
||
|
||
The signal pipeline (engine + aggregator + backtester) consumes ONLY the
|
||
public contract defined here, so swapping providers never touches signal
|
||
code.
|
||
"""
|
||
|
||
import logging
|
||
import threading
|
||
import time
|
||
from abc import ABC, abstractmethod
|
||
from datetime import datetime, timedelta
|
||
from typing import AsyncIterator, Callable, Dict, Optional, Sequence, Tuple
|
||
|
||
import pandas as pd
|
||
|
||
from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD
|
||
|
||
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 BaseDataProvider(ABC):
|
||
"""Shared OHLCV provider behavior; subclasses supply the raw fetch."""
|
||
|
||
#: Registry name; overridden per provider.
|
||
name: str = "base"
|
||
|
||
#: Exception type this provider raises for fetch/validation failures.
|
||
#: Provider-specific errors subclass DataIntegrityError so consumers
|
||
#: can catch one type regardless of the active provider.
|
||
DATA_ERROR = DataIntegrityError
|
||
|
||
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,
|
||
}
|
||
|
||
#: Hard per-request bar cap shared by Twelve Data and Oanda.
|
||
MAX_BARS_PER_REQUEST = 5000
|
||
|
||
# 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._cache: Dict[str, pd.DataFrame] = {}
|
||
self._cache_timestamps: Dict[str, float] = {}
|
||
# The bot shares one provider across a thread pool; cache eviction
|
||
# is check-then-act and needs locking.
|
||
self._cache_lock = threading.Lock()
|
||
|
||
# ── Abstract: the only provider-specific piece ─────────────────
|
||
|
||
@abstractmethod
|
||
def _fetch_raw(
|
||
self,
|
||
symbol: str,
|
||
timeframe: Timeframe,
|
||
outputsize: int,
|
||
start: Optional[datetime] = None,
|
||
end: Optional[datetime] = None,
|
||
) -> pd.DataFrame:
|
||
"""Fetch up to `outputsize` bars, optionally bounded by start/end
|
||
(tz-aware UTC datetimes). Must return a normalized frame:
|
||
capitalized OHLCV columns, UTC DatetimeIndex named "Datetime",
|
||
oldest-first. Raise self.DATA_ERROR on failure."""
|
||
|
||
# ── 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,
|
||
self.MAX_BARS_PER_REQUEST),
|
||
)
|
||
df = self._fetch_raw(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)
|
||
try:
|
||
retry = self._fetch_raw(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
|
||
except DataIntegrityError as e:
|
||
logger.warning(
|
||
f"{symbol} {timeframe.value}: boundary retry refetch "
|
||
f"failed ({e}) — keeping original data")
|
||
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 fetch_range(
|
||
self,
|
||
symbol: str,
|
||
timeframe: Timeframe,
|
||
start: datetime,
|
||
end: datetime,
|
||
) -> pd.DataFrame:
|
||
"""Fetch the RAW closed-bar series between start/end, paginating
|
||
under the per-request bar cap. Returns uncleaned bars (the
|
||
backtester resolves trades on the raw price path); callers wanting
|
||
the signal-side series run `validate_and_clean` on the result."""
|
||
bars_per_day = self.BARS_PER_DAY[timeframe]
|
||
days_per_chunk = max(1, int(4500 / bars_per_day))
|
||
|
||
chunks = []
|
||
cursor_end = end
|
||
empty_streak = 0
|
||
while cursor_end > start:
|
||
cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk))
|
||
try:
|
||
chunk = self._fetch_raw(
|
||
symbol, timeframe, self.MAX_BARS_PER_REQUEST,
|
||
start=cursor_start, end=cursor_end,
|
||
)
|
||
except Exception as e:
|
||
# Providers raise for windows with no data (weekends,
|
||
# holidays, asset gaps). Skip the window, step back, retry
|
||
# — but bail after a few consecutive empties.
|
||
logger.warning(
|
||
f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}"
|
||
)
|
||
empty_streak += 1
|
||
if empty_streak >= 3:
|
||
break
|
||
cursor_end = cursor_start - timedelta(seconds=1)
|
||
continue
|
||
|
||
if chunk is None or chunk.empty:
|
||
empty_streak += 1
|
||
if empty_streak >= 3:
|
||
break
|
||
cursor_end = cursor_start - timedelta(seconds=1)
|
||
continue
|
||
|
||
empty_streak = 0
|
||
chunks.append(chunk)
|
||
# Step back; dedup on concat.
|
||
cursor_end = chunk.index[0] - timedelta(seconds=1)
|
||
if len(chunk) < 2:
|
||
break
|
||
|
||
if not chunks:
|
||
raise RuntimeError(f"No data fetched for {symbol} {timeframe.value}")
|
||
|
||
df = pd.concat(chunks).sort_index()
|
||
df = df[~df.index.duplicated(keep="last")]
|
||
# Fetches ending "now" include the still-forming candle — drop it
|
||
# so callers only ever see closed bars.
|
||
df = self._drop_forming_candle(df, timeframe)
|
||
start_ts = pd.Timestamp(start)
|
||
end_ts = pd.Timestamp(end)
|
||
return df[(df.index >= start_ts) & (df.index <= end_ts)]
|
||
|
||
def is_data_fresh(
|
||
self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
|
||
) -> bool:
|
||
"""Public freshness check (last bar within 3 candle durations)."""
|
||
return self._is_data_fresh(df, symbol, timeframe)
|
||
|
||
def validate_and_clean(
|
||
self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
|
||
) -> pd.DataFrame:
|
||
"""Public entry to the shared cleaning pipeline."""
|
||
return self._validate_and_clean(df, symbol, timeframe)
|
||
|
||
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}
|
||
|
||
def get_spread(self, symbol: str = XAUUSD) -> float:
|
||
"""Latest observed round-trip spread; providers without live
|
||
bid/ask fall back to the pair's assumed default."""
|
||
return PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])["default_spread"]
|
||
|
||
# ── Streaming-quote capability (optional) ──────────────────────
|
||
|
||
#: True when the provider can stream live ticks via `iter_ticks`.
|
||
supports_streaming: bool = False
|
||
|
||
def iter_ticks(
|
||
self,
|
||
symbols: Sequence[str],
|
||
on_connect: Optional[Callable[[], None]] = None,
|
||
) -> AsyncIterator[Tuple[str, float]]:
|
||
"""Async generator yielding (canonical_symbol, price) ticks for one
|
||
connection's lifetime; the caller owns reconnect/backoff. Providers
|
||
that support it set `supports_streaming = True` and override."""
|
||
raise NotImplementedError(
|
||
f"{self.name} provider does not support streaming quotes")
|
||
|
||
# ── Shared internals ───────────────────────────────────────────
|
||
|
||
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.
|
||
"""
|
||
if df is None or df.empty:
|
||
return False
|
||
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.
|
||
|
||
Some feeds include 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. Feeds that only
|
||
return complete candles pass through unchanged.
|
||
"""
|
||
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 _validate_and_clean(
|
||
self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
|
||
) -> pd.DataFrame:
|
||
if df is None or df.empty:
|
||
raise self.DATA_ERROR(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
|