From e6ef6e8c9d0bc6993ea34356275616b82aea9e6d Mon Sep 17 00:00:00 2001 From: Harwinder Date: Mon, 4 May 2026 17:48:20 -0400 Subject: [PATCH] Remove synthetic data fallback; fix concurrent API credit exhaustion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit data_provider: - Remove _generate_synthetic_data entirely — no more fake prices - Raise DataUnavailableError instead of falling back to synthetic data (no API key, daily limit hit, API returns no values, fetch failure) - Move rate limiter to module-level globals (_rate_lock, _rate_request_times, _rate_daily_credits) so all ForexDataProvider instances running in parallel threads share the same 8-credit/min and 800-credit/day accounting - _throttle now raises DataUnavailableError on daily limit (was silent return False) - Remove _no_api_data tracking, numpy/hashlib imports telegram_bot: - Detect F-grade reports caused by data unavailability and show clean "Data unavailable" line per TF instead of a garbled zero-price card Co-Authored-By: Claude Sonnet 4.6 --- data_provider.py | 267 ++++++++++++++++------------------------------- telegram_bot.py | 13 +++ 2 files changed, 102 insertions(+), 178 deletions(-) diff --git a/data_provider.py b/data_provider.py index 2e3574b..3c787b9 100644 --- a/data_provider.py +++ b/data_provider.py @@ -2,18 +2,18 @@ Forex Signal Intelligence Platform - Data Provider ==================================================== Retrieves and validates market data from Twelve Data API. -Falls back to synthetic GBM data if the API is unavailable. +Raises DataUnavailableError when the API is unreachable or credits are exhausted. +No synthetic/simulated data fallback. """ -import hashlib import logging import os +import threading import time from collections import deque from datetime import datetime, timedelta, timezone from typing import Dict, List, Optional -import numpy as np import pandas as pd import requests @@ -27,6 +27,20 @@ class DataIntegrityError(Exception): pass +class DataUnavailableError(Exception): + """Raised when live market data cannot be fetched (no key, rate limit, API error).""" + pass + + +# ── Module-level shared rate limiter ─────────────────────────────── +# All ForexDataProvider instances share these so concurrent scans +# don't each think they have a full 8-credit budget. +_rate_lock = threading.Lock() +_rate_request_times: deque = deque() +_rate_daily_credits: int = 0 +_rate_daily_reset: float = 0.0 + + class ForexDataProvider: """ Forex data provider backed by Twelve Data API. @@ -73,12 +87,7 @@ class ForexDataProvider: ) self._cache: Dict[str, pd.DataFrame] = {} self._cache_timestamps: Dict[str, float] = {} - self._request_times: deque = deque() # timestamps of recent per-minute credits - self._daily_credits: int = 0 # credits used today - self._daily_reset: float = time.time() + 86400 # next midnight reset - self._no_api_data: set = set() # td_symbols with no values from API (use synthetic) self._session = requests.Session() - # Docs recommend Authorization header over apikey query param self._session.headers.update({ "User-Agent": "Mozilla/5.0 (compatible; ForexIntel/1.0)", }) @@ -207,54 +216,66 @@ class ForexDataProvider: # ── Private Methods ──────────────────────────────────────────── - def _throttle(self, credits: int = 1) -> bool: + def _throttle(self, credits: int = 1) -> None: """ Block until `credits` per-minute API credits are available. - Returns False if the daily limit is reached (caller should use synthetic data). + Raises DataUnavailableError if the daily limit is exhausted. + + Uses module-level state so all ForexDataProvider instances (including + those running in parallel threads) share the same credit accounting. Twelve Data free tier: 8 credits/minute, 800 credits/day. - Each symbol in any request (single or batch) consumes 1 credit. """ - # Reset daily counter after 24 hours - now = time.time() - if now >= self._daily_reset: - self._daily_credits = 0 - self._daily_reset = now + 86400 + global _rate_daily_credits, _rate_daily_reset - # Daily limit check - if self._daily_credits + credits > self.DAILY_LIMIT: - logger.warning( - f"Daily API limit reached ({self._daily_credits}/{self.DAILY_LIMIT} credits) " - "— using cached/synthetic data for remaining requests today." - ) - return False - - # Warn when approaching daily limit - if self._daily_credits + credits > self.DAILY_LIMIT * 0.8: - logger.warning( - f"Approaching daily API limit: {self._daily_credits}/{self.DAILY_LIMIT} credits used." - ) - - # Per-minute rate limit - while True: + with _rate_lock: 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: - ts = time.time() - for _ in range(credits): - self._request_times.append(ts) - self._daily_credits += credits - return True + # Reset daily counter after 24 hours + if _rate_daily_reset == 0.0: + _rate_daily_reset = now + 86400 + if now >= _rate_daily_reset: + _rate_daily_credits = 0 + _rate_daily_reset = now + 86400 - 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 for next window..." - ) - time.sleep(wait) + # Daily limit check — raise immediately, no fallback + if _rate_daily_credits + credits > self.DAILY_LIMIT: + raise DataUnavailableError( + f"Twelve Data daily API limit reached " + f"({_rate_daily_credits}/{self.DAILY_LIMIT} credits). " + "Resets in 24 hours." + ) + + if _rate_daily_credits + credits > self.DAILY_LIMIT * 0.8: + logger.warning( + f"Approaching daily API limit: " + f"{_rate_daily_credits}/{self.DAILY_LIMIT} credits used." + ) + + # Per-minute rate limit — wait inside the lock so other threads + # queue up rather than over-counting available credits. + while True: + now = time.time() + while _rate_request_times and now - _rate_request_times[0] >= self.RATE_WINDOW: + _rate_request_times.popleft() + + available = self.RATE_LIMIT - len(_rate_request_times) + if available >= credits: + ts = time.time() + for _ in range(credits): + _rate_request_times.append(ts) + _rate_daily_credits += credits + return + + wait = (_rate_request_times[0] + self.RATE_WINDOW) - now + 0.2 + logger.info( + f"Rate limit: {len(_rate_request_times)}/{self.RATE_LIMIT} credits used " + f"— waiting {wait:.1f}s for next window…" + ) + # Release lock while sleeping so other threads can check state + _rate_lock.release() + time.sleep(wait) + _rate_lock.acquire() def _batch_fetch_timeframe( self, symbols: List[str], timeframe: Timeframe, lookback_days: int @@ -274,9 +295,9 @@ class ForexDataProvider: ) outputsize = max(outputsize, 300) - # Each symbol in a batch request costs 1 credit - if not self._throttle(credits=len(symbols)): - return # daily limit reached — callers will use synthetic data + # Each symbol in a batch request costs 1 credit — raises DataUnavailableError + # if daily limit reached; per-minute waits are handled inside _throttle. + self._throttle(credits=len(symbols)) resp = self._session.get( f"{self.TD_BASE_URL}/time_series", @@ -311,20 +332,10 @@ class ForexDataProvider: values = sym_data.get("values") if not values: - if td_sym not in self._no_api_data: - logger.info( - f"No API data for {td_sym} — using synthetic data" - ) - self._no_api_data.add(td_sym) - # Cache synthetic so fetch_ohlcv won't make a redundant API call - try: - syn = self._generate_synthetic_data(symbol, timeframe, lookback_days) - syn = self._validate_and_clean(syn, symbol, timeframe) - cache_key = f"{symbol}_{timeframe.value}" - self._cache[cache_key] = syn - self._cache_timestamps[cache_key] = now - except Exception: - pass + logger.warning( + f"No data returned by Twelve Data for {td_sym} " + f"— symbol may require a paid plan." + ) continue records = [ @@ -356,12 +367,12 @@ class ForexDataProvider: ) -> pd.DataFrame: """ Fetch OHLCV from Twelve Data REST API. - Falls back to synthetic GBM data if API is unavailable or key is missing. + Raises DataUnavailableError when data cannot be retrieved. """ if not self._api_key: - logger.warning("No Twelve Data API key — using synthetic data. " - "Set TWELVE_DATA_API_KEY env var or config.twelve_data_api_key.") - return self._generate_synthetic_data(symbol, timeframe, lookback_days) + raise DataUnavailableError( + "No Twelve Data API key — set the TWELVE_DATA_API_KEY environment variable." + ) td_symbol = self._to_td_symbol(symbol) interval = self.TD_INTERVALS[timeframe] @@ -371,8 +382,8 @@ class ForexDataProvider: ) outputsize = max(outputsize, 300) - if not self._throttle(credits=1): - return self._generate_synthetic_data(symbol, timeframe, lookback_days) + # Raises DataUnavailableError on daily limit; blocks on per-minute limit. + self._throttle(credits=1) try: resp = self._session.get( @@ -390,16 +401,16 @@ class ForexDataProvider: data = resp.json() if data.get("status") == "error": - raise DataIntegrityError( + raise DataUnavailableError( f"Twelve Data API error for {td_symbol}: {data.get('message', 'unknown')}" ) values = data.get("values") if not values: - raise DataIntegrityError(f"Empty response from Twelve Data for {td_symbol}") + raise DataUnavailableError( + f"No data returned for {td_symbol} — symbol may require a paid plan." + ) - # Values arrive newest-first; reverse to chronological order - # API may return null for any OHLC field — map to None for pandas NaN handling records = [ { "Datetime": pd.Timestamp(v["datetime"], tz="UTC"), @@ -418,112 +429,12 @@ class ForexDataProvider: logger.debug(f"Fetched {len(df)} bars for {td_symbol} @ {interval}") return df - except DataIntegrityError: + except (DataUnavailableError, DataIntegrityError): raise except Exception as e: - logger.info(f"Twelve Data fetch failed for {symbol} ({e}), using synthetic data.") - return self._generate_synthetic_data(symbol, timeframe, lookback_days) - - def _generate_synthetic_data( - self, symbol: str, timeframe: Timeframe, lookback_days: int - ) -> pd.DataFrame: - """ - Generate realistic synthetic Forex OHLCV using GBM with mean-reversion. - Used as a fallback when live data is unavailable. - """ - pair_params = { - "EURUSD=X": {"price": 1.1350, "vol": 0.07, "drift": -0.001}, - "GBPUSD=X": {"price": 1.3200, "vol": 0.08, "drift": 0.002}, - "USDJPY=X": {"price": 144.50, "vol": 0.09, "drift": 0.003}, - "USDCHF=X": {"price": 0.8850, "vol": 0.07, "drift": -0.001}, - "AUDUSD=X": {"price": 0.6550, "vol": 0.10, "drift": -0.002}, - "USDCAD=X": {"price": 1.3650, "vol": 0.07, "drift": 0.001}, - "NZDUSD=X": {"price": 0.6100, "vol": 0.10, "drift": -0.001}, - "EURGBP=X": {"price": 0.8590, "vol": 0.06, "drift": 0.000}, - "EURJPY=X": {"price": 163.90, "vol": 0.09, "drift": 0.002}, - "GBPJPY=X": {"price": 190.70, "vol": 0.11, "drift": 0.003}, - "AUDJPY=X": {"price": 94.60, "vol": 0.11, "drift": 0.001}, - "EURCHF=X": {"price": 0.9380, "vol": 0.05, "drift": 0.000}, - "CADJPY=X": {"price": 105.80, "vol": 0.09, "drift": 0.001}, - } - - params = pair_params.get(symbol, {"price": 1.0000, "vol": 0.08, "drift": 0.000}) - base_price = params["price"] - annual_vol = params["vol"] - annual_drift = params["drift"] - - tf_minutes = { - Timeframe.M1: 1, Timeframe.M5: 5, Timeframe.M15: 15, - Timeframe.M30: 30, Timeframe.H1: 60, Timeframe.H4: 240, - Timeframe.D1: 1440, Timeframe.W1: 10080, - } - bar_minutes = tf_minutes.get(timeframe, 60) - bars_per_year = (365.25 * 24 * 60) / bar_minutes - dt = 1.0 / bars_per_year - - n_bars = min(int(lookback_days * 24 * 60 / bar_minutes), 2000) - n_bars = max(n_bars, 300) - - seed_base = int(hashlib.md5(symbol.encode()).hexdigest()[:8], 16) - time_component = int(time.time()) // 3600 - np.random.seed((seed_base + time_component) % (2**31)) - - bar_vol = annual_vol * np.sqrt(dt) - bar_drift = annual_drift * dt - - regime_len = np.random.randint(20, 80) - trend_strength = np.random.normal(0, 0.3) - - price = base_price - prices = [price] - - for i in range(1, n_bars): - if i % regime_len == 0: - regime_len = np.random.randint(20, 80) - trend_strength = np.random.normal(0, 0.3) - - mr_force = -0.001 * (price - base_price) / base_price - ret = bar_drift + trend_strength * dt + mr_force + bar_vol * np.random.randn() - - if np.random.random() < 0.02: - ret *= np.random.uniform(2.0, 4.0) - - price *= (1 + ret) - prices.append(price) - - prices = np.array(prices) - opens = np.roll(prices, 1) - opens[0] = prices[0] - - typical_range = bar_vol * base_price * 1.5 - highs = np.maximum(prices, opens) + np.abs(np.random.normal(0, typical_range * 0.6, n_bars)) - lows = np.minimum(prices, opens) - np.abs(np.random.normal(0, typical_range * 0.6, n_bars)) - highs = np.maximum(highs, np.maximum(opens, prices)) - lows = np.minimum(lows, np.minimum(opens, prices)) - - volume = np.random.lognormal(mean=10, sigma=0.5, size=n_bars).astype(int) - - freq_map = { - Timeframe.M1: "1min", Timeframe.M5: "5min", - Timeframe.M15: "15min", Timeframe.M30: "30min", - Timeframe.H1: "1h", Timeframe.H4: "4h", - Timeframe.D1: "D", Timeframe.W1: "W", - } - freq = freq_map.get(timeframe, "1h") - end_time = pd.Timestamp.now(tz="UTC").floor(f"{bar_minutes}min") - index = pd.date_range(end=end_time, periods=n_bars, freq=freq, tz="UTC") - - df = pd.DataFrame({ - "Open": opens, "High": highs, "Low": lows, - "Close": prices, "Volume": volume, - }, index=index) - df.index.name = "Datetime" - - logger.info( - f"Generated {n_bars} synthetic {timeframe.value} bars for " - f"{SYMBOL_DISPLAY.get(symbol, symbol)} @ {prices[-1]:.5f}" - ) - return df + raise DataUnavailableError( + f"Failed to fetch {td_symbol}: {e}" + ) from e def _validate_and_clean( self, df: pd.DataFrame, symbol: str, timeframe: Timeframe diff --git a/telegram_bot.py b/telegram_bot.py index 2b20402..63753cd 100644 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -137,6 +137,19 @@ def _build_combined_signal( continue r = reports[0] + # If primary TF data was unavailable the engine returns grade F with a warning. + # Show a clean "data unavailable" line instead of a garbled card. + if r.signal_quality == "F" and any( + "unavailable" in w.lower() or "api" in w.lower() or "credits" in w.lower() + for w in r.warnings + ): + lines.append("") + lines.append("────────────────────────") + lines.append(f"{tf} — ❌ Data unavailable") + if r.warnings: + lines.append(f"{_h(r.warnings[0])}") + continue + dot = _grade_dot(r.signal_quality) action = _signal_action(r.final_signal)