Harwinder
7ccd8c4b04
Remove backtest.py, backtest_gpu.py, metrics.py, local CSV data, Dukascopy loader, and all backtest CLI modes from main.py. Rebalance engine weights: leading indicators (divergence, RSI, stochastic, BB) now carry 66% of signal weight vs 34% for lagging (EMA, MACD). Add ws_twelvedata.py for live tick streaming. Drop pyarrow and tqdm deps.
266 lines
8.9 KiB
Python
266 lines
8.9 KiB
Python
"""
|
|
ForexBot2 — Data Provider (Twelve Data SDK, USD/JPY only)
|
|
==========================================================
|
|
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, DISPLAY, PIP_VALUE
|
|
|
|
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",
|
|
}
|
|
|
|
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:
|
|
"""Estimate spread from recent M5 H-L median; fallback to 1 pip on JPY."""
|
|
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
|
|
# Fallback: 1 pip = 0.01 for JPY pairs
|
|
return PIP_VALUE
|
|
|
|
def get_pair_info(self, symbol: str = USDJPY) -> Dict:
|
|
return {
|
|
"symbol": symbol,
|
|
"display": DISPLAY,
|
|
"pip_value": PIP_VALUE,
|
|
"is_jpy_pair": True,
|
|
}
|
|
|
|
# ── 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
|