diff --git a/data_provider.py b/data_provider.py index 8e1cbbe..da28e95 100644 --- a/data_provider.py +++ b/data_provider.py @@ -17,11 +17,35 @@ from typing import Dict, List, Optional import pandas as pd import requests +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__) _no_data_warned: set = set() # suppress duplicate "no data" warnings per symbol +# Symbols that Twelve Data free tier does not support — route to Yahoo Finance. +# Gold/silver futures are the most reliable Yahoo symbols (full OHLCV history). +_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.D1: "1d", + Timeframe.W1: "1wk", +} + class DataIntegrityError(Exception): """Raised when data fails validation checks.""" @@ -227,6 +251,57 @@ class ForexDataProvider: # ── Private Methods ──────────────────────────────────────────── + def _fetch_yfinance_fallback( + self, symbol: str, timeframe: Timeframe + ) -> Optional[pd.DataFrame]: + """ + Fetch OHLCV from Yahoo Finance for symbols Twelve Data doesn't cover + on the free tier (XAUUSD, XAGUSD). Returns None if not mapped or fails. + """ + if not _YF_AVAILABLE: + return None + yf_symbol = _YF_FALLBACK_MAP.get(symbol) + yf_interval = _YF_INTERVALS.get(timeframe) + if yf_symbol is None or yf_interval is None: + return None + + # Yahoo intraday limits: 1m → last 7d, 15m/30m/1h → last 60d + if yf_interval in ("1m", "2m"): + period = "7d" + elif yf_interval in ("5m", "15m", "30m", "60m", "1h"): + period = "60d" + else: + period = "2y" + + try: + df = yf.download( + yf_symbol, interval=yf_interval, period=period, + progress=False, auto_adjust=False, threads=False, + ) + if df is None or df.empty: + return None + + # Single-symbol download can return MultiIndex columns in newer yfinance versions + if isinstance(df.columns, pd.MultiIndex): + df.columns = df.columns.get_level_values(0) + + # Normalise to UTC tz-aware index + 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() + logger.info( + f"[yf] {yf_symbol} {yf_interval}: {len(df)} bars, " + f"last close={df['Close'].iloc[-1]:.5f}" + ) + return df + except Exception as e: + logger.warning(f"yfinance fallback failed for {symbol} {timeframe.value}: {e}") + return None + def _throttle(self, credits: int = 1) -> None: """ Block until `credits` per-minute API credits are available. @@ -343,6 +418,14 @@ class ForexDataProvider: values = sym_data.get("values") if not values: + # Try yfinance fallback for unsupported symbols (XAU, XAG) + yf_df = self._fetch_yfinance_fallback(symbol, timeframe) + if yf_df is not None: + yf_df = self._validate_and_clean(yf_df, symbol, timeframe) + cache_key = f"{symbol}_{timeframe.value}" + self._cache[cache_key] = yf_df + self._cache_timestamps[cache_key] = now + continue if td_sym not in _no_data_warned: logger.warning(f"No data from Twelve Data for {td_sym} — symbol may require a paid plan.") _no_data_warned.add(td_sym) @@ -378,8 +461,13 @@ class ForexDataProvider: """ Fetch OHLCV from Twelve Data REST API. Raises DataUnavailableError when data cannot be retrieved. + Falls back to yfinance for symbols not on the Twelve Data free tier. """ if not self._api_key: + # No Twelve Data key — try yfinance fallback for mapped symbols + yf_df = self._fetch_yfinance_fallback(symbol, timeframe) + if yf_df is not None: + return yf_df raise DataUnavailableError( "No Twelve Data API key — set the TWELVE_DATA_API_KEY environment variable." ) @@ -417,6 +505,9 @@ class ForexDataProvider: values = data.get("values") if not values: + yf_df = self._fetch_yfinance_fallback(symbol, timeframe) + if yf_df is not None: + return yf_df raise DataUnavailableError( f"No data returned for {td_symbol} — symbol may require a paid plan." ) diff --git a/requirements.txt b/requirements.txt index 615386f..85efc56 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ numpy>=1.24 pandas>=2.0 requests>=2.28 +yfinance>=0.2.40 python-telegram-bot[job-queue]>=20.0 diff --git a/telegram_bot.py b/telegram_bot.py index bb55221..ba3894c 100644 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -276,7 +276,7 @@ class ForexTelegramBot: return SignalAggregator(self._build_config(tf, pairs), data_provider=provider).analyze_all_pairs() async def _analyze_all_tfs( - self, pairs: List[str] + self, pairs: List[str], progress=None, ) -> Dict[str, List[SignalReport]]: """Single combined prefetch, then run all SCAN_TIMEFRAMES concurrently off the warm cache.""" # Collect all unique timeframes needed across every TF scan @@ -286,9 +286,16 @@ class ForexTelegramBot: for tf in [_TF_MAP[tf_name]] + _TF_CONFIRMATIONS.get(tf_name, []) }) provider = ForexDataProvider(PlatformConfig()) - provider.prefetch_batch(list(pairs), all_tfs) loop = asyncio.get_running_loop() + if progress: + await progress(f"📡 Fetching market data ({len(all_tfs)} timeframes × {len(pairs)} pairs)…") + # Run blocking prefetch off the event loop so we don't freeze the bot + await loop.run_in_executor(None, provider.prefetch_batch, list(pairs), all_tfs) + + if progress: + await progress("🧠 Analyzing pairs across timeframes…") + tasks = { tf: loop.run_in_executor(self._executor, self._run_scan, tf, list(pairs), provider) for tf in SCAN_TIMEFRAMES @@ -368,8 +375,15 @@ class ForexTelegramBot: msg = await update.message.reply_text( f"🔄 Analyzing {_h(display)} across {_h(tfs_str)}…" ) + + async def progress(text: str): + try: + await msg.edit_text(text) + except Exception: + pass # ignore edit-rate errors / no-change errors + try: - results_by_tf = await self._analyze_all_tfs([raw]) + results_by_tf = await self._analyze_all_tfs([raw], progress=progress) now_str = _now_est() # Check we got at least one result @@ -396,8 +410,15 @@ class ForexTelegramBot: msg = await update.message.reply_text( f"🔄 Scanning all pairs across {_h(tfs_str)}…" ) + + async def progress(text: str): + try: + await msg.edit_text(text) + except Exception: + pass + try: - results_by_tf = await self._analyze_all_tfs(list(MAJOR_PAIRS)) + results_by_tf = await self._analyze_all_tfs(list(MAJOR_PAIRS), progress=progress) now_str = _now_est() # Best grade per pair across all TFs