Harwinder
404 lines
14 KiB
Python
404 lines
14 KiB
Python
"""
|
|
Forex Signal Intelligence Platform - Data Provider
|
|
====================================================
|
|
Retrieves and validates market data from Yahoo Finance API.
|
|
Raises DataUnavailableError when the API is unreachable.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import time
|
|
from collections import deque
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Dict, List, Optional
|
|
|
|
import pandas as pd
|
|
|
|
try:
|
|
import yfinance as yf
|
|
_YF_AVAILABLE = True
|
|
except ImportError:
|
|
yf = None
|
|
_YF_AVAILABLE = False
|
|
|
|
from config import PlatformConfig, Timeframe, SYMBOL_DISPLAY, PIP_VALUES
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Fallbacks for specific symbols to get accurate Forex/Commodity data from yfinance.
|
|
_YF_FALLBACK_MAP = {
|
|
"XAUUSD=X": "GC=F",
|
|
"XAGUSD=X": "SI=F",
|
|
}
|
|
|
|
_YF_INTERVALS = {
|
|
Timeframe.M1: "1m",
|
|
Timeframe.M5: "5m",
|
|
Timeframe.M15: "15m",
|
|
Timeframe.M30: "30m",
|
|
Timeframe.H1: "1h",
|
|
Timeframe.H4: "1h", # Resampled in code
|
|
Timeframe.D1: "1d",
|
|
Timeframe.W1: "1wk",
|
|
}
|
|
|
|
class DataIntegrityError(Exception):
|
|
"""Raised when data fails validation checks."""
|
|
pass
|
|
|
|
class DataUnavailableError(Exception):
|
|
"""Raised when live market data cannot be fetched (API error)."""
|
|
pass
|
|
|
|
class ForexDataProvider:
|
|
"""
|
|
Forex data provider backed by yfinance API.
|
|
"""
|
|
|
|
def __init__(self, config: PlatformConfig):
|
|
self.config = config
|
|
self._cache: Dict[str, pd.DataFrame] = {}
|
|
self._cache_timestamps: Dict[str, float] = {}
|
|
|
|
def _cache_ttl(self, timeframe: Timeframe) -> int:
|
|
"""Cache TTL for a given timeframe — falls back to refresh_interval."""
|
|
return self.config.cache_ttl_per_tf.get(timeframe, self.config.refresh_interval)
|
|
|
|
def fetch_ohlcv(
|
|
self,
|
|
symbol: str,
|
|
timeframe: Timeframe,
|
|
lookback_days: Optional[int] = None,
|
|
) -> pd.DataFrame:
|
|
"""
|
|
Fetch OHLCV data for a Forex pair with full validation.
|
|
|
|
Returns a DataFrame with columns: Open, High, Low, Close, Volume
|
|
Index is a timezone-aware DatetimeIndex in UTC.
|
|
"""
|
|
if lookback_days is None:
|
|
lookback_days = self.config.data_lookback_days
|
|
|
|
cache_key = f"{symbol}_{timeframe.value}"
|
|
now = time.time()
|
|
ttl = self._cache_ttl(timeframe)
|
|
|
|
if (
|
|
cache_key in self._cache_timestamps
|
|
and (now - self._cache_timestamps[cache_key]) < ttl
|
|
and cache_key in self._cache
|
|
):
|
|
cached = self._cache.get(cache_key)
|
|
if cached is not None and not cached.empty:
|
|
return cached
|
|
|
|
df = self._fetch_yfinance(symbol, timeframe, lookback_days)
|
|
df = self._validate_and_clean(df, symbol, timeframe)
|
|
|
|
self._cache[cache_key] = df
|
|
self._cache_timestamps[cache_key] = now
|
|
return df
|
|
|
|
def prefetch_batch(
|
|
self,
|
|
symbols: List[str],
|
|
timeframes: List[Timeframe],
|
|
lookback_days: Optional[int] = None,
|
|
) -> None:
|
|
"""
|
|
Batch-fetch all symbols for each timeframe in a single API call per TF.
|
|
"""
|
|
if lookback_days is None:
|
|
lookback_days = self.config.data_lookback_days
|
|
|
|
now = time.time()
|
|
for tf in timeframes:
|
|
ttl = self._cache_ttl(tf)
|
|
cached = all(
|
|
f"{s}_{tf.value}" in self._cache_timestamps
|
|
and (now - self._cache_timestamps[f"{s}_{tf.value}"]) < ttl
|
|
and not self._cache.get(f"{s}_{tf.value}", pd.DataFrame()).empty
|
|
for s in symbols
|
|
)
|
|
if cached:
|
|
logger.debug(f"Skipping prefetch for {tf.value} — all {len(symbols)} symbols cached")
|
|
continue
|
|
try:
|
|
self._batch_fetch_timeframe(symbols, tf, lookback_days)
|
|
except Exception as e:
|
|
logger.warning(f"Batch prefetch failed for {tf.value}: {e}")
|
|
|
|
def fetch_multiple(
|
|
self,
|
|
symbols: List[str],
|
|
timeframe: Timeframe,
|
|
lookback_days: Optional[int] = None,
|
|
) -> Dict[str, pd.DataFrame]:
|
|
"""Fetch data for multiple symbols, returning dict of DataFrames."""
|
|
results = {}
|
|
for symbol in symbols:
|
|
try:
|
|
df = self.fetch_ohlcv(symbol, timeframe, lookback_days)
|
|
results[symbol] = df
|
|
except Exception as e:
|
|
logger.warning(f"Failed to fetch {symbol}: {e}")
|
|
return results
|
|
|
|
def get_latest_price(self, symbol: str) -> Optional[float]:
|
|
"""Return the most recent close price for a symbol."""
|
|
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) -> float:
|
|
"""Estimate typical spread from recent M5 high-low ranges."""
|
|
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_spreads = {
|
|
"EURUSD=X": 0.00010, "GBPUSD=X": 0.00015,
|
|
"USDJPY=X": 0.010, "USDCHF=X": 0.00015,
|
|
"AUDUSD=X": 0.00015, "USDCAD=X": 0.00018,
|
|
"NZDUSD=X": 0.00020,
|
|
}
|
|
return fallback_spreads.get(symbol, 0.00020)
|
|
|
|
def get_pair_info(self, symbol: str) -> Dict:
|
|
"""Return display name and pip value for a pair."""
|
|
display = SYMBOL_DISPLAY.get(symbol, symbol)
|
|
is_jpy = "JPY" in symbol
|
|
pip_value = PIP_VALUES.get(symbol, 0.01 if is_jpy else 0.0001)
|
|
return {
|
|
"symbol": symbol,
|
|
"display": display,
|
|
"pip_value": pip_value,
|
|
"is_jpy_pair": is_jpy,
|
|
}
|
|
|
|
# ── Private Methods ────────────────────────────────────────────
|
|
|
|
def _fetch_yfinance(self, symbol: str, timeframe: Timeframe, lookback_days: int) -> pd.DataFrame:
|
|
if not _YF_AVAILABLE:
|
|
raise DataUnavailableError("yfinance is not installed.")
|
|
|
|
yf_symbol = _YF_FALLBACK_MAP.get(symbol, symbol)
|
|
|
|
# Determine actual interval and resampling needs
|
|
actual_timeframe = timeframe
|
|
resample_needed = False
|
|
if timeframe == Timeframe.H4:
|
|
actual_timeframe = Timeframe.H1
|
|
resample_needed = True
|
|
|
|
yf_interval = _YF_INTERVALS.get(actual_timeframe)
|
|
if yf_interval is None:
|
|
raise DataUnavailableError(f"Unsupported timeframe {timeframe.value} for yfinance")
|
|
|
|
# Period logic based on Yahoo limitations
|
|
if yf_interval in ("1m", "2m", "5m"):
|
|
period = "7d" if lookback_days < 7 else "60d"
|
|
elif yf_interval in ("15m", "30m", "1h", "60m"):
|
|
period = "60d"
|
|
elif yf_interval in ("1d", "1wk"):
|
|
period = "2y"
|
|
else:
|
|
period = f"{lookback_days}d"
|
|
|
|
for attempt in range(3):
|
|
try:
|
|
df = yf.download(
|
|
yf_symbol, interval=yf_interval, period=period,
|
|
progress=False, auto_adjust=False, threads=False,
|
|
)
|
|
if df is not None and not df.empty:
|
|
break
|
|
time.sleep(1.0)
|
|
except Exception as e:
|
|
if attempt == 2:
|
|
raise DataUnavailableError(f"yfinance fetch failed for {symbol} {timeframe.value}: {e}")
|
|
time.sleep(1.0)
|
|
else:
|
|
raise DataUnavailableError(f"No data returned for {symbol}")
|
|
|
|
if isinstance(df.columns, pd.MultiIndex):
|
|
df.columns = df.columns.get_level_values(0)
|
|
|
|
if df.index.tz is None:
|
|
df.index = df.index.tz_localize("UTC")
|
|
else:
|
|
df.index = df.index.tz_convert("UTC")
|
|
df.index.name = "Datetime"
|
|
df = df[["Open", "High", "Low", "Close", "Volume"]].copy()
|
|
|
|
if resample_needed and timeframe == Timeframe.H4:
|
|
# Resample 1h data into 4h blocks
|
|
df = df.resample('4h').agg({
|
|
'Open': 'first',
|
|
'High': 'max',
|
|
'Low': 'min',
|
|
'Close': 'last',
|
|
'Volume': 'sum'
|
|
}).dropna()
|
|
|
|
logger.info(f"[yf live] {yf_symbol} {timeframe.value}: {len(df)} bars, last close={df['Close'].iloc[-1]:.5f}")
|
|
return df
|
|
|
|
def _batch_fetch_timeframe(self, symbols: List[str], timeframe: Timeframe, lookback_days: int) -> None:
|
|
if not _YF_AVAILABLE:
|
|
return
|
|
|
|
actual_timeframe = timeframe
|
|
resample_needed = False
|
|
if timeframe == Timeframe.H4:
|
|
actual_timeframe = Timeframe.H1
|
|
resample_needed = True
|
|
|
|
yf_interval = _YF_INTERVALS.get(actual_timeframe)
|
|
if not yf_interval:
|
|
return
|
|
|
|
yf_symbols = [_YF_FALLBACK_MAP.get(s, s) for s in symbols]
|
|
|
|
if yf_interval in ("1m", "2m", "5m"):
|
|
period = "7d" if lookback_days < 7 else "60d"
|
|
elif yf_interval in ("15m", "30m", "1h", "60m"):
|
|
period = "60d"
|
|
elif yf_interval in ("1d", "1wk"):
|
|
period = "2y"
|
|
else:
|
|
period = f"{lookback_days}d"
|
|
|
|
for attempt in range(3):
|
|
try:
|
|
df_multi = yf.download(
|
|
" ".join(yf_symbols), interval=yf_interval, period=period,
|
|
progress=False, auto_adjust=False, threads=False, group_by="ticker"
|
|
)
|
|
if df_multi is not None and not df_multi.empty:
|
|
break
|
|
time.sleep(1.0)
|
|
except Exception as e:
|
|
logger.warning(f"Batch fetch attempt {attempt+1} failed for {timeframe.value}: {e}")
|
|
time.sleep(1.0)
|
|
else:
|
|
logger.warning(f"Batch fetch empty for {timeframe.value} after 3 attempts")
|
|
return
|
|
|
|
if df_multi.index.tz is None:
|
|
df_multi.index = df_multi.index.tz_localize("UTC")
|
|
else:
|
|
df_multi.index = df_multi.index.tz_convert("UTC")
|
|
df_multi.index.name = "Datetime"
|
|
|
|
now = time.time()
|
|
for symbol, yf_sym in zip(symbols, yf_symbols):
|
|
try:
|
|
if len(yf_symbols) == 1:
|
|
df_sym = df_multi.copy()
|
|
if isinstance(df_sym.columns, pd.MultiIndex):
|
|
df_sym.columns = df_sym.columns.get_level_values(1)
|
|
else:
|
|
if yf_sym in df_multi.columns.get_level_values(0):
|
|
df_sym = df_multi[yf_sym].copy()
|
|
else:
|
|
# Might be single level columns if download only returned 1 valid ticker
|
|
if yf_sym == yf_symbols[0] and not isinstance(df_multi.columns, pd.MultiIndex):
|
|
df_sym = df_multi.copy()
|
|
else:
|
|
continue
|
|
|
|
# Ignore if column names are not Open/High/Low/Close
|
|
if 'Close' not in df_sym.columns:
|
|
continue
|
|
|
|
df_sym = df_sym[["Open", "High", "Low", "Close", "Volume"]].copy()
|
|
|
|
if resample_needed and timeframe == Timeframe.H4:
|
|
df_sym = df_sym.resample('4h').agg({
|
|
'Open': 'first',
|
|
'High': 'max',
|
|
'Low': 'min',
|
|
'Close': 'last',
|
|
'Volume': 'sum'
|
|
}).dropna()
|
|
|
|
df_sym = self._validate_and_clean(df_sym, symbol, timeframe)
|
|
cache_key = f"{symbol}_{timeframe.value}"
|
|
self._cache[cache_key] = df_sym
|
|
self._cache_timestamps[cache_key] = now
|
|
logger.info(f"[yf batch] {yf_sym} {timeframe.value}: {len(df_sym)} bars, last close={df_sym['Close'].iloc[-1]:.5f}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to process batch data for {symbol}: {e}")
|
|
|
|
def _validate_and_clean(
|
|
self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
|
|
) -> pd.DataFrame:
|
|
"""
|
|
Validate and clean OHLCV data.
|
|
"""
|
|
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.debug(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.debug(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.debug(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.debug(
|
|
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.debug(f"{symbol}: Cleaned {cleaned}/{original_len} bars")
|
|
|
|
return df
|