Harwinder
a708609f8b
Add EURUSD pair with dynamic pair info lookup. Telegram bot now shows 4 scan buttons: EUR/USD 15M, EUR/USD 1H, USD/JPY 15M, USD/JPY 1H. Each indicator displays directional arrows in the message output. Price formatting adapts per pair (5dp for EUR, 3dp for JPY).
267 lines
8.9 KiB
Python
267 lines
8.9 KiB
Python
"""
|
|
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,
|
|
}
|
|
|
|
RATE_LIMIT = 8
|
|
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:
|
|
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()
|
|
|
|
if (
|
|
cache_key in self._cache_timestamps
|
|
and (now - self._cache_timestamps[cache_key]) < self.config.refresh_interval
|
|
and cache_key in self._cache
|
|
):
|
|
cached = self._cache.get(cache_key)
|
|
if cached is not None and not cached.empty:
|
|
return cached
|
|
|
|
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)
|
|
|
|
self._cache[cache_key] = df
|
|
self._cache_timestamps[cache_key] = now
|
|
return df
|
|
|
|
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:
|
|
"""Free tier: 8 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"])
|
|
|
|
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,
|
|
self.config.technical.ichimoku_senkou_b,
|
|
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."
|
|
)
|
|
|
|
df = df.sort_index()
|
|
df = df[~df.index.duplicated(keep="last")]
|
|
|
|
cleaned = original_len - len(df)
|
|
if cleaned > 0:
|
|
logger.info(f"{symbol}: Cleaned {cleaned}/{original_len} bars")
|
|
|
|
return df
|