Add yfinance fallback for gold/silver + coarse Telegram progress
- yfinance>=0.2.40 added as a hard dependency
- _fetch_yfinance_fallback(): pulls OHLCV from Yahoo for symbols not on
the Twelve Data free tier (XAUUSD → GC=F, XAGUSD → SI=F)
- Hooked into both batch and single-symbol fetch paths; logs as [yf]
to distinguish from Twelve Data [live]
- Fail-loud: if both providers return nothing, the pair is skipped with
the existing "data unavailable" message
- _analyze_all_tfs accepts an async progress callback; both /signal and
/scan now show coarse milestones via message edits:
"📡 Fetching market data..." → "🧠 Analyzing pairs..." → result
- prefetch_batch is now run in a thread so it doesn't block the event loop
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -17,11 +17,35 @@ from typing import Dict, List, Optional
|
|||||||
import pandas as pd
|
import pandas as pd
|
||||||
import requests
|
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
|
from config import PlatformConfig, Timeframe, SYMBOL_DISPLAY, PIP_VALUES
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
_no_data_warned: set = set() # suppress duplicate "no data" warnings per symbol
|
_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):
|
class DataIntegrityError(Exception):
|
||||||
"""Raised when data fails validation checks."""
|
"""Raised when data fails validation checks."""
|
||||||
@@ -227,6 +251,57 @@ class ForexDataProvider:
|
|||||||
|
|
||||||
# ── Private Methods ────────────────────────────────────────────
|
# ── 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:
|
def _throttle(self, credits: int = 1) -> None:
|
||||||
"""
|
"""
|
||||||
Block until `credits` per-minute API credits are available.
|
Block until `credits` per-minute API credits are available.
|
||||||
@@ -343,6 +418,14 @@ class ForexDataProvider:
|
|||||||
|
|
||||||
values = sym_data.get("values")
|
values = sym_data.get("values")
|
||||||
if not 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:
|
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.")
|
logger.warning(f"No data from Twelve Data for {td_sym} — symbol may require a paid plan.")
|
||||||
_no_data_warned.add(td_sym)
|
_no_data_warned.add(td_sym)
|
||||||
@@ -378,8 +461,13 @@ class ForexDataProvider:
|
|||||||
"""
|
"""
|
||||||
Fetch OHLCV from Twelve Data REST API.
|
Fetch OHLCV from Twelve Data REST API.
|
||||||
Raises DataUnavailableError when data cannot be retrieved.
|
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:
|
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(
|
raise DataUnavailableError(
|
||||||
"No Twelve Data API key — set the TWELVE_DATA_API_KEY environment variable."
|
"No Twelve Data API key — set the TWELVE_DATA_API_KEY environment variable."
|
||||||
)
|
)
|
||||||
@@ -417,6 +505,9 @@ class ForexDataProvider:
|
|||||||
|
|
||||||
values = data.get("values")
|
values = data.get("values")
|
||||||
if not values:
|
if not values:
|
||||||
|
yf_df = self._fetch_yfinance_fallback(symbol, timeframe)
|
||||||
|
if yf_df is not None:
|
||||||
|
return yf_df
|
||||||
raise DataUnavailableError(
|
raise DataUnavailableError(
|
||||||
f"No data returned for {td_symbol} — symbol may require a paid plan."
|
f"No data returned for {td_symbol} — symbol may require a paid plan."
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
numpy>=1.24
|
numpy>=1.24
|
||||||
pandas>=2.0
|
pandas>=2.0
|
||||||
requests>=2.28
|
requests>=2.28
|
||||||
|
yfinance>=0.2.40
|
||||||
python-telegram-bot[job-queue]>=20.0
|
python-telegram-bot[job-queue]>=20.0
|
||||||
|
|||||||
+25
-4
@@ -276,7 +276,7 @@ class ForexTelegramBot:
|
|||||||
return SignalAggregator(self._build_config(tf, pairs), data_provider=provider).analyze_all_pairs()
|
return SignalAggregator(self._build_config(tf, pairs), data_provider=provider).analyze_all_pairs()
|
||||||
|
|
||||||
async def _analyze_all_tfs(
|
async def _analyze_all_tfs(
|
||||||
self, pairs: List[str]
|
self, pairs: List[str], progress=None,
|
||||||
) -> Dict[str, List[SignalReport]]:
|
) -> Dict[str, List[SignalReport]]:
|
||||||
"""Single combined prefetch, then run all SCAN_TIMEFRAMES concurrently off the warm cache."""
|
"""Single combined prefetch, then run all SCAN_TIMEFRAMES concurrently off the warm cache."""
|
||||||
# Collect all unique timeframes needed across every TF scan
|
# 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, [])
|
for tf in [_TF_MAP[tf_name]] + _TF_CONFIRMATIONS.get(tf_name, [])
|
||||||
})
|
})
|
||||||
provider = ForexDataProvider(PlatformConfig())
|
provider = ForexDataProvider(PlatformConfig())
|
||||||
provider.prefetch_batch(list(pairs), all_tfs)
|
|
||||||
|
|
||||||
loop = asyncio.get_running_loop()
|
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 = {
|
tasks = {
|
||||||
tf: loop.run_in_executor(self._executor, self._run_scan, tf, list(pairs), provider)
|
tf: loop.run_in_executor(self._executor, self._run_scan, tf, list(pairs), provider)
|
||||||
for tf in SCAN_TIMEFRAMES
|
for tf in SCAN_TIMEFRAMES
|
||||||
@@ -368,8 +375,15 @@ class ForexTelegramBot:
|
|||||||
msg = await update.message.reply_text(
|
msg = await update.message.reply_text(
|
||||||
f"🔄 Analyzing {_h(display)} across {_h(tfs_str)}…"
|
f"🔄 Analyzing {_h(display)} across {_h(tfs_str)}…"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def progress(text: str):
|
||||||
try:
|
try:
|
||||||
results_by_tf = await self._analyze_all_tfs([raw])
|
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], progress=progress)
|
||||||
now_str = _now_est()
|
now_str = _now_est()
|
||||||
|
|
||||||
# Check we got at least one result
|
# Check we got at least one result
|
||||||
@@ -396,8 +410,15 @@ class ForexTelegramBot:
|
|||||||
msg = await update.message.reply_text(
|
msg = await update.message.reply_text(
|
||||||
f"🔄 Scanning all pairs across {_h(tfs_str)}…"
|
f"🔄 Scanning all pairs across {_h(tfs_str)}…"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def progress(text: str):
|
||||||
try:
|
try:
|
||||||
results_by_tf = await self._analyze_all_tfs(list(MAJOR_PAIRS))
|
await msg.edit_text(text)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
try:
|
||||||
|
results_by_tf = await self._analyze_all_tfs(list(MAJOR_PAIRS), progress=progress)
|
||||||
now_str = _now_est()
|
now_str = _now_est()
|
||||||
|
|
||||||
# Best grade per pair across all TFs
|
# Best grade per pair across all TFs
|
||||||
|
|||||||
Reference in New Issue
Block a user