From df77e0fb5ebcc9f871909c8ce507dc1bd1b86ed5 Mon Sep 17 00:00:00 2001 From: Harwinder Date: Mon, 4 May 2026 20:48:25 -0400 Subject: [PATCH] =?UTF-8?q?Fix=20concurrent=20API=20credit=20exhaustion=20?= =?UTF-8?q?=E2=80=94=20shared=20prefetch=20+=20dedup=20warnings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - telegram_bot: one ForexDataProvider shared across all SCAN_TIMEFRAMES; combined prefetch of all unique TFs up front so the 4 concurrent scans hit the warm cache instead of firing simultaneous API calls - data_provider: skip prefetch_batch for timeframes already in cache - data_provider: deduplicate "no data" warning per symbol (fires once, not once per TF) - data_provider: [live] info log on successful fetch; outlier-removal downgraded to debug Co-Authored-By: Claude Sonnet 4.6 --- data_provider.py | 18 ++++++++++++++---- signal_aggregator.py | 4 ++-- telegram_bot.py | 18 ++++++++++++++---- 3 files changed, 30 insertions(+), 10 deletions(-) diff --git a/data_provider.py b/data_provider.py index c469eec..8e1cbbe 100644 --- a/data_provider.py +++ b/data_provider.py @@ -20,6 +20,7 @@ import requests 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 class DataIntegrityError(Exception): @@ -151,7 +152,17 @@ class ForexDataProvider: if lookback_days is None: lookback_days = self.config.data_lookback_days + now = time.time() for tf in timeframes: + cached = all( + f"{s}_{tf.value}" in self._cache_timestamps + and (now - self._cache_timestamps[f"{s}_{tf.value}"]) < self.config.refresh_interval + 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) time.sleep(1.0) # brief pause between TF requests @@ -332,10 +343,9 @@ class ForexDataProvider: values = sym_data.get("values") if not values: - logger.warning( - f"No data returned by Twelve Data for {td_sym} " - f"— symbol may require a paid plan." - ) + 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) continue records = [ diff --git a/signal_aggregator.py b/signal_aggregator.py index 38122bb..7b07e3c 100644 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -104,9 +104,9 @@ class SignalAggregator: fundamental analysis, and produces final graded signals. """ - def __init__(self, config: PlatformConfig): + def __init__(self, config: PlatformConfig, data_provider: Optional[ForexDataProvider] = None): self.config = config - self.data_provider = ForexDataProvider(config) + self.data_provider = data_provider or ForexDataProvider(config) self.tech_engine = TechnicalEngine(config.technical) self.news_engine = NewsEngine(config.news) diff --git a/telegram_bot.py b/telegram_bot.py index 63753cd..4673a8c 100644 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -33,6 +33,7 @@ from config import ( MAJOR_PAIRS, CROSS_PAIRS, SYMBOL_DISPLAY, PlatformConfig, SignalStrength, Timeframe, ) +from data_provider import ForexDataProvider from signal_aggregator import SignalAggregator, SignalReport logger = logging.getLogger(__name__) @@ -282,16 +283,25 @@ class ForexTelegramBot: cfg.watch_pairs = list(pairs) return cfg - def _run_scan(self, tf: str, pairs: List[str]) -> List[SignalReport]: - return SignalAggregator(self._build_config(tf, pairs)).analyze_all_pairs() + def _run_scan(self, tf: str, pairs: List[str], provider: ForexDataProvider) -> List[SignalReport]: + return SignalAggregator(self._build_config(tf, pairs), data_provider=provider).analyze_all_pairs() async def _analyze_all_tfs( self, pairs: List[str] ) -> Dict[str, List[SignalReport]]: - """Run all SCAN_TIMEFRAMES concurrently, return {tf: [reports]}.""" + """Single combined prefetch, then run all SCAN_TIMEFRAMES concurrently off the warm cache.""" + # Collect all unique timeframes needed across every TF scan + all_tfs = list({ + tf + for tf_name in SCAN_TIMEFRAMES + 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() tasks = { - tf: loop.run_in_executor(self._executor, self._run_scan, tf, list(pairs)) + tf: loop.run_in_executor(self._executor, self._run_scan, tf, list(pairs), provider) for tf in SCAN_TIMEFRAMES } results: Dict[str, List[SignalReport]] = {}