harwinder
39c91c1aa4
- Move index dedup before OHLC consistency fix in _validate_and_clean to prevent crash when df.loc[dup_idx] returns Series instead of scalar - Add global flood backoff: detect RetryAfter, pause all sends/edits until ban expires instead of retrying and escalating the ban - Stop dashboard edit-fail → new-message spam cascade: track consecutive failures, only resend after 3 failed edits - Increase dashboard refresh from 5s to 30s, min send interval to 1.5s - Add comprehensive QA test suite (140 tests across all 7 modules)
317 lines
11 KiB
Python
Executable File
317 lines
11 KiB
Python
Executable File
"""
|
|
ForexBot2 — Data Provider (Twelve Data SDK)
|
|
==========================================================
|
|
Fetches and validates OHLCV via the official `twelvedata` Python SDK.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import time
|
|
from collections import deque
|
|
from typing import Dict, Optional
|
|
|
|
import pandas as pd
|
|
from twelvedata import TDClient
|
|
|
|
from config import PlatformConfig, Timeframe, USDJPY, PAIR_INFO
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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 = (
|
|
os.environ.get("TWELVE_DATA_API_KEY")
|
|
or config.twelve_data_api_key
|
|
)
|
|
self._cache: Dict[str, pd.DataFrame] = {}
|
|
self._cache_timestamps: Dict[str, float] = {}
|
|
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 = USDJPY,
|
|
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)
|
|
if (
|
|
cache_key in self._cache_timestamps
|
|
and (now - self._cache_timestamps[cache_key]) < cache_ttl
|
|
and cache_key in self._cache
|
|
):
|
|
cached = self._cache.get(cache_key)
|
|
if cached is not None and not cached.empty:
|
|
if self._is_data_fresh(cached, symbol, timeframe):
|
|
return cached
|
|
logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching")
|
|
del self._cache[cache_key]
|
|
del self._cache_timestamps[cache_key]
|
|
|
|
outputsize = max(
|
|
300,
|
|
min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, 5000),
|
|
)
|
|
df = self._fetch_twelve_data(symbol, timeframe, outputsize)
|
|
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."
|
|
)
|
|
|
|
self._cache[cache_key] = df
|
|
self._cache_timestamps[cache_key] = now
|
|
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 now.weekday() >= 5:
|
|
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 = USDJPY) -> 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 = USDJPY) -> 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[USDJPY])
|
|
return info["pip_value"]
|
|
|
|
def get_pair_info(self, symbol: str = USDJPY) -> Dict:
|
|
info = PAIR_INFO.get(symbol, PAIR_INFO[USDJPY])
|
|
return {"symbol": symbol, **info}
|
|
|
|
# ── Private ────────────────────────────────────────────────────
|
|
|
|
def _throttle(self, credits: int = 1) -> None:
|
|
"""Pro tier: 55 credits / minute. Block until budget allows."""
|
|
while True:
|
|
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
|
|
|
|
wait = (self._request_times[0] + self.RATE_WINDOW) - now + 0.2
|
|
logger.info(
|
|
f"Rate limit: {len(self._request_times)}/{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 Twelve Data API key. Set TWELVE_DATA_API_KEY env var "
|
|
"or config.twelve_data_api_key."
|
|
)
|
|
|
|
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:
|
|
outlier_mask = (returns - returns.mean()).abs() > 10 * std_ret
|
|
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
|