Remove synthetic data fallback; fix concurrent API credit exhaustion

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 <noreply@anthropic.com>
This commit is contained in:
Harwinder
2026-05-04 17:48:20 -04:00
parent c4e1ba5589
commit e6ef6e8c9d
2 changed files with 102 additions and 178 deletions
+77 -166
View File
@@ -2,18 +2,18 @@
Forex Signal Intelligence Platform - Data Provider Forex Signal Intelligence Platform - Data Provider
==================================================== ====================================================
Retrieves and validates market data from Twelve Data API. 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 logging
import os import os
import threading
import time import time
from collections import deque from collections import deque
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional from typing import Dict, List, Optional
import numpy as np
import pandas as pd import pandas as pd
import requests import requests
@@ -27,6 +27,20 @@ class DataIntegrityError(Exception):
pass 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: class ForexDataProvider:
""" """
Forex data provider backed by Twelve Data API. Forex data provider backed by Twelve Data API.
@@ -73,12 +87,7 @@ class ForexDataProvider:
) )
self._cache: Dict[str, pd.DataFrame] = {} self._cache: Dict[str, pd.DataFrame] = {}
self._cache_timestamps: Dict[str, float] = {} 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() self._session = requests.Session()
# Docs recommend Authorization header over apikey query param
self._session.headers.update({ self._session.headers.update({
"User-Agent": "Mozilla/5.0 (compatible; ForexIntel/1.0)", "User-Agent": "Mozilla/5.0 (compatible; ForexIntel/1.0)",
}) })
@@ -207,54 +216,66 @@ class ForexDataProvider:
# ── Private Methods ──────────────────────────────────────────── # ── 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. 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. 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 global _rate_daily_credits, _rate_daily_reset
with _rate_lock:
now = time.time() now = time.time()
if now >= self._daily_reset:
self._daily_credits = 0
self._daily_reset = now + 86400
# Daily limit check # Reset daily counter after 24 hours
if self._daily_credits + credits > self.DAILY_LIMIT: if _rate_daily_reset == 0.0:
logger.warning( _rate_daily_reset = now + 86400
f"Daily API limit reached ({self._daily_credits}/{self.DAILY_LIMIT} credits) " if now >= _rate_daily_reset:
"— using cached/synthetic data for remaining requests today." _rate_daily_credits = 0
) _rate_daily_reset = now + 86400
return False
# Warn when approaching daily limit # Daily limit check — raise immediately, no fallback
if self._daily_credits + credits > self.DAILY_LIMIT * 0.8: if _rate_daily_credits + credits > self.DAILY_LIMIT:
logger.warning( raise DataUnavailableError(
f"Approaching daily API limit: {self._daily_credits}/{self.DAILY_LIMIT} credits used." f"Twelve Data daily API limit reached "
f"({_rate_daily_credits}/{self.DAILY_LIMIT} credits). "
"Resets in 24 hours."
) )
# Per-minute rate limit 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: while True:
now = time.time() now = time.time()
while self._request_times and now - self._request_times[0] >= self.RATE_WINDOW: while _rate_request_times and now - _rate_request_times[0] >= self.RATE_WINDOW:
self._request_times.popleft() _rate_request_times.popleft()
available = self.RATE_LIMIT - len(self._request_times) available = self.RATE_LIMIT - len(_rate_request_times)
if available >= credits: if available >= credits:
ts = time.time() ts = time.time()
for _ in range(credits): for _ in range(credits):
self._request_times.append(ts) _rate_request_times.append(ts)
self._daily_credits += credits _rate_daily_credits += credits
return True return
wait = (self._request_times[0] + self.RATE_WINDOW) - now + 0.2 wait = (_rate_request_times[0] + self.RATE_WINDOW) - now + 0.2
logger.info( logger.info(
f"Rate limit: {len(self._request_times)}/{self.RATE_LIMIT} credits used " f"Rate limit: {len(_rate_request_times)}/{self.RATE_LIMIT} credits used "
f"— waiting {wait:.1f}s for next window..." f"— waiting {wait:.1f}s for next window…"
) )
# Release lock while sleeping so other threads can check state
_rate_lock.release()
time.sleep(wait) time.sleep(wait)
_rate_lock.acquire()
def _batch_fetch_timeframe( def _batch_fetch_timeframe(
self, symbols: List[str], timeframe: Timeframe, lookback_days: int self, symbols: List[str], timeframe: Timeframe, lookback_days: int
@@ -274,9 +295,9 @@ class ForexDataProvider:
) )
outputsize = max(outputsize, 300) outputsize = max(outputsize, 300)
# Each symbol in a batch request costs 1 credit # Each symbol in a batch request costs 1 credit — raises DataUnavailableError
if not self._throttle(credits=len(symbols)): # if daily limit reached; per-minute waits are handled inside _throttle.
return # daily limit reached — callers will use synthetic data self._throttle(credits=len(symbols))
resp = self._session.get( resp = self._session.get(
f"{self.TD_BASE_URL}/time_series", f"{self.TD_BASE_URL}/time_series",
@@ -311,20 +332,10 @@ class ForexDataProvider:
values = sym_data.get("values") values = sym_data.get("values")
if not values: if not values:
if td_sym not in self._no_api_data: logger.warning(
logger.info( f"No data returned by Twelve Data for {td_sym} "
f"No API data for {td_sym} — using synthetic data" f"— symbol may require a paid plan."
) )
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
continue continue
records = [ records = [
@@ -356,12 +367,12 @@ class ForexDataProvider:
) -> pd.DataFrame: ) -> pd.DataFrame:
""" """
Fetch OHLCV from Twelve Data REST API. 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: if not self._api_key:
logger.warning("No Twelve Data API key — using synthetic data. " raise DataUnavailableError(
"Set TWELVE_DATA_API_KEY env var or config.twelve_data_api_key.") "No Twelve Data API key — set the TWELVE_DATA_API_KEY environment variable."
return self._generate_synthetic_data(symbol, timeframe, lookback_days) )
td_symbol = self._to_td_symbol(symbol) td_symbol = self._to_td_symbol(symbol)
interval = self.TD_INTERVALS[timeframe] interval = self.TD_INTERVALS[timeframe]
@@ -371,8 +382,8 @@ class ForexDataProvider:
) )
outputsize = max(outputsize, 300) outputsize = max(outputsize, 300)
if not self._throttle(credits=1): # Raises DataUnavailableError on daily limit; blocks on per-minute limit.
return self._generate_synthetic_data(symbol, timeframe, lookback_days) self._throttle(credits=1)
try: try:
resp = self._session.get( resp = self._session.get(
@@ -390,16 +401,16 @@ class ForexDataProvider:
data = resp.json() data = resp.json()
if data.get("status") == "error": if data.get("status") == "error":
raise DataIntegrityError( raise DataUnavailableError(
f"Twelve Data API error for {td_symbol}: {data.get('message', 'unknown')}" f"Twelve Data API error for {td_symbol}: {data.get('message', 'unknown')}"
) )
values = data.get("values") values = data.get("values")
if not 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 = [ records = [
{ {
"Datetime": pd.Timestamp(v["datetime"], tz="UTC"), "Datetime": pd.Timestamp(v["datetime"], tz="UTC"),
@@ -418,112 +429,12 @@ class ForexDataProvider:
logger.debug(f"Fetched {len(df)} bars for {td_symbol} @ {interval}") logger.debug(f"Fetched {len(df)} bars for {td_symbol} @ {interval}")
return df return df
except DataIntegrityError: except (DataUnavailableError, DataIntegrityError):
raise raise
except Exception as e: except Exception as e:
logger.info(f"Twelve Data fetch failed for {symbol} ({e}), using synthetic data.") raise DataUnavailableError(
return self._generate_synthetic_data(symbol, timeframe, lookback_days) f"Failed to fetch {td_symbol}: {e}"
) from e
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
def _validate_and_clean( def _validate_and_clean(
self, df: pd.DataFrame, symbol: str, timeframe: Timeframe self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
+13
View File
@@ -137,6 +137,19 @@ def _build_combined_signal(
continue continue
r = reports[0] 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"<b>{tf}</b> — ❌ Data unavailable")
if r.warnings:
lines.append(f"<i>{_h(r.warnings[0])}</i>")
continue
dot = _grade_dot(r.signal_quality) dot = _grade_dot(r.signal_quality)
action = _signal_action(r.final_signal) action = _signal_action(r.final_signal)