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)