diff --git a/.claude/settings.json b/.claude/settings.json new file mode 100644 index 0000000..1b02605 --- /dev/null +++ b/.claude/settings.json @@ -0,0 +1,43 @@ +{ + "permissions": { + "allow": [ + "Bash(python -c \"from config import PlatformConfig; from engine import TechnicalEngine; from data_provider import ForexDataProvider; from news_engine import NewsEngine; from signal_aggregator import SignalAggregator, SignalReport; print\\('All imports OK'\\)\")", + "Bash(pip install *)", + "Bash(python -c ' *)", + "Bash(python -m venv .venv)", + "Bash(.venv\\\\Scripts\\\\python.exe -m pip install pandas numpy requests yfinance python-dateutil -q)", + "PowerShell(& \".venv\\\\Scripts\\\\python.exe\" -m pip install pandas numpy requests yfinance python-dateutil -q 2>&1)", + "PowerShell(& \".venv\\\\Scripts\\\\python.exe\" main.py 2>&1)", + "PowerShell(& \".venv\\\\Scripts\\\\python.exe\" -c \"from data_provider import ForexDataProvider; print\\('import OK'\\)\")", + "Bash(dir h:\\\\Apps\\\\htdocs\\\\ForexBot2\\\\*.py /b)", + "Bash(.venv/Scripts/pip freeze *)", + "Bash(git add *)", + "Bash(git push *)", + "Bash(git pull *)", + "Bash(git checkout *)", + "Bash(git rebase *)", + "Bash(Get-ChildItem -Recurse -File)", + "Bash(Select-Object FullName)", + "Bash(cd /d \"h:\\\\Apps\\\\htdocs\\\\ForexBot2\")", + "Bash(python -c \"import ast; ast.parse\\(open\\('telegram_bot.py'\\).read\\(\\)\\); print\\('Syntax OK'\\)\")", + "Bash(python -c \"import ast; ast.parse\\(open\\(r'h:/Apps/htdocs/ForexBot2/telegram_bot.py'\\).read\\(\\)\\); print\\('Syntax OK'\\)\")", + "Bash(python -c \"import ast; ast.parse\\(open\\(r'h:/Apps/htdocs/ForexBot2/telegram_bot.py', encoding='utf-8'\\).read\\(\\)\\); print\\('Syntax OK'\\)\")", + "Bash(git -C \"h:/Apps/htdocs/ForexBot2\" status)", + "Bash(git -C \"h:/Apps/htdocs/ForexBot2\" diff)", + "Bash(git -C \"h:/Apps/htdocs/ForexBot2\" log --oneline -5)", + "Bash(git -C \"h:/Apps/htdocs/ForexBot2\" add telegram_bot.py)", + "Bash(git -C h:/Apps/htdocs/ForexBot2 commit -m ' *)", + "Bash(git -C \"h:/Apps/htdocs/ForexBot2\" push)", + "Bash(git -C \"h:/Apps/htdocs/ForexBot2\" remote set-url origin http://harry:Sekhon007@192.168.2.10:3000/harry/ForexBot2.git)", + "Bash(git -C \"h:/Apps/htdocs/ForexBot2\" add USAGE.md)", + "Bash(ssh forex2@192.168.2.10 \"pkill -f telegram_bot.py; sleep 1; pgrep -fa telegram_bot.py || echo 'No bot processes running'\")", + "Bash(python -c \"import ast; ast.parse\\(open\\(r'h:/Apps/htdocs/ForexBot2/telegram_bot.py', encoding='utf-8'\\).read\\(\\)\\); print\\('OK'\\)\")", + "Bash(git -C \"h:/Apps/htdocs/ForexBot2\" add CLAUDE.md)", + "Bash(git -C \"h:/Apps/htdocs/ForexBot2\" add data_provider.py telegram_bot.py)", + "WebFetch(domain:github.com)", + "WebFetch(domain:finnhub.io)", + "Bash(python -c \"import ast; ast.parse\\(open\\('data_provider_finnhub.py'\\).read\\(\\)\\); ast.parse\\(open\\('signal_aggregator.py'\\).read\\(\\)\\); ast.parse\\(open\\('telegram_bot.py'\\).read\\(\\)\\); ast.parse\\(open\\('config.py'\\).read\\(\\)\\); print\\('All files parse OK'\\)\")", + "Bash(python -c \"import io, ast; ast.parse\\(io.open\\('data_provider_finnhub.py', encoding='utf-8'\\).read\\(\\)\\); print\\('OK'\\)\")" + ] + } +} diff --git a/CLAUDE.md b/CLAUDE.md index 432f714..e9161a8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,64 +1,21 @@ # ForexBot2 — Known Bugs & Dev Notes -## Bug 1 — Concurrent TF scans exhaust API credits instantly +## [RESOLVED] Bug 1 — Concurrent TF scans exhaust API credits instantly -**Symptom:** -``` -Batch prefetch failed for 1wk: You have run out of API credits for the current minute. -9 API credits were used, with the current limit at https://twelvedata.com/pricing -``` -Warnings appear for every timeframe (1h, 4h, 1d, 1wk) in rapid succession. - -**Root cause:** -`telegram_bot.py` runs all 4 `SCAN_TIMEFRAMES` concurrently via `ThreadPoolExecutor`. -Each `SignalAggregator` fires its own batch prefetch simultaneously. -Free tier limit is **8 credits/minute** — 4 parallel scans hit this in under a second. - -**Impact:** -Confirmation timeframes fail and fall back to synthetic data. -Primary TF still has real data (first request wins the credits). - -**Fix (not yet implemented):** -Run timeframes sequentially with a 60-second wait between each, or reduce -`SCAN_TIMEFRAMES` to a single primary TF and use the engine's built-in -multi-timeframe confirmation (which already batches efficiently in 3 calls). +**Status:** Resolved +**Resolution:** Replaced the Twelve Data API with `yfinance` in `data_provider.py`. Since `yfinance` has no strict per-minute rate limiting, concurrent scans no longer fail. Furthermore, the `signal_aggregator.py` has been updated to use `ThreadPoolExecutor` to speed up pair analysis in parallel. --- -## Bug 2 — XAUUSD (Gold) shows wrong price (~1.01 instead of ~3300) +## [RESOLVED] Bug 2 — XAUUSD (Gold) shows wrong price (~1.01 instead of ~3300) -**Symptom:** -``` -Entry: 1.01037 -Take Profit: 1.00767 (+0.3 pips) -Stop Loss: 1.01172 (-0.1 pips) -``` - -**Root cause:** -Two compounding issues: - -1. XAUUSD is not available on the Twelve Data **free tier** — API returns no data. -2. `data_provider._generate_synthetic_data()` uses a `pair_params` dict for base - prices. XAUUSD either falls through to a wrong default (~1.0) instead of the - correct ~3300 baseline. - -Check `data_provider.py` → `_generate_synthetic_data()` → `pair_params` dict. -Verify `"XAUUSD=X"` has a correct `"price"` entry around `3300.0`. - -**Impact:** -All indicator values, entry, TP, and SL for XAUUSD are meaningless when -running on the free tier (synthetic data with wrong base price). - -**Fix (not yet implemented):** -- Correct the base price in `pair_params` for `XAUUSD=X`. -- Add a bot-level guard: if `report.current_price < 100` for XAUUSD, - suppress the trade setup and show a "data unavailable" message instead. +**Status:** Resolved +**Resolution:** By transitioning entirely to `yfinance`, XAUUSD is now natively mapped to `GC=F` via the `_YF_FALLBACK_MAP`. This means we get the real, correct OHLCV data directly without needing synthetic fallbacks. --- ## Notes -- Free tier: 8 credits/min, 800 credits/day. Each full scan = ~21 credits. -- Supported pairs (live data on free tier): EURUSD, GBPUSD, USDJPY, USDCHF, - AUDUSD, USDCAD, NZDUSD, EURGBP, EURJPY, GBPJPY, AUDJPY, EURCHF, CADJPY. -- All other symbols (XAU, XAG, commodities, indices) fall back to synthetic GBM data. +- **Data Provider**: The primary data provider is now `yfinance`. `Twelve Data` has been completely removed along with all its throttle logic and API Key requirements. +- **Parallelization**: Pair analysis is now multithreaded using `ThreadPoolExecutor` in `signal_aggregator.py`, maximizing the speed of data fetching and processing. +- **Supported pairs**: All forex pairs (EURUSD, GBPUSD, USDJPY, etc.) are fetched with an `=X` suffix in Yahoo Finance. Commodities like Gold (XAUUSD=X) use their futures symbols (`GC=F`). diff --git a/__pycache__/config.cpython-314.pyc b/__pycache__/config.cpython-314.pyc index 886d755..5809deb 100644 Binary files a/__pycache__/config.cpython-314.pyc and b/__pycache__/config.cpython-314.pyc differ diff --git a/__pycache__/data_provider.cpython-314.pyc b/__pycache__/data_provider.cpython-314.pyc index 61a6bc3..930dda6 100644 Binary files a/__pycache__/data_provider.cpython-314.pyc and b/__pycache__/data_provider.cpython-314.pyc differ diff --git a/__pycache__/signal_aggregator.cpython-314.pyc b/__pycache__/signal_aggregator.cpython-314.pyc index 1770de2..908f214 100644 Binary files a/__pycache__/signal_aggregator.cpython-314.pyc and b/__pycache__/signal_aggregator.cpython-314.pyc differ diff --git a/config.py b/config.py index 955ad2e..0a934e1 100644 --- a/config.py +++ b/config.py @@ -193,8 +193,7 @@ class PlatformConfig: # Data source data_lookback_days: int = 60 # enough for 200 EMA on H1 - # Twelve Data API key (overridden by TWELVE_DATA_API_KEY env var) - twelve_data_api_key: str = "f563b22ffdfd4924955decbe1a8c8701" + # Dashboard refresh interval (seconds) — used as a fallback when a # timeframe isn't covered by cache_ttl_per_tf below. diff --git a/data_provider.py b/data_provider.py index aa24073..86c2a57 100644 --- a/data_provider.py +++ b/data_provider.py @@ -1,21 +1,18 @@ """ Forex Signal Intelligence Platform - Data Provider ==================================================== -Retrieves and validates market data from Twelve Data API. -Raises DataUnavailableError when the API is unreachable or credits are exhausted. -No synthetic/simulated data fallback. +Retrieves and validates market data from Yahoo Finance API. +Raises DataUnavailableError when the API is unreachable. """ 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 pandas as pd -import requests try: import yfinance as yf @@ -27,10 +24,8 @@ except ImportError: 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 -# 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). +# Fallbacks for specific symbols to get accurate Forex/Commodity data from yfinance. _YF_FALLBACK_MAP = { "XAUUSD=X": "GC=F", "XAGUSD=X": "SI=F", @@ -42,88 +37,28 @@ _YF_INTERVALS = { Timeframe.M15: "15m", Timeframe.M30: "30m", Timeframe.H1: "1h", + Timeframe.H4: "1h", # Resampled in code Timeframe.D1: "1d", Timeframe.W1: "1wk", } - class DataIntegrityError(Exception): """Raised when data fails validation checks.""" pass - class DataUnavailableError(Exception): - """Raised when live market data cannot be fetched (no key, rate limit, API error).""" + """Raised when live market data cannot be fetched (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. - - Twelve Data has native 4h candles, eliminating the H1→H4 resampling - hack needed for Yahoo Finance. Free tier: 800 credits/day, 8 req/min. + Forex data provider backed by yfinance API. """ - TD_BASE_URL = "https://api.twelvedata.com" - - TD_INTERVALS = { - Timeframe.M1: "1min", - Timeframe.M5: "5min", - Timeframe.M15: "15min", - Timeframe.M30: "30min", - Timeframe.H1: "1h", - Timeframe.H4: "4h", - Timeframe.D1: "1day", - Timeframe.W1: "1week", - } - - # Approximate bars per day for each timeframe (used to size outputsize) - BARS_PER_DAY = { - Timeframe.M1: 1440, - Timeframe.M5: 288, - Timeframe.M15: 96, - Timeframe.M30: 48, - Timeframe.H1: 24, - Timeframe.H4: 6, - Timeframe.D1: 1, - Timeframe.W1: 0.143, - } - - # Free plan: 8 credits/minute, 800 credits/day - RATE_LIMIT = 8 - RATE_WINDOW = 60.0 - DAILY_LIMIT = 800 - def __init__(self, config: PlatformConfig): self.config = config - self._api_key = ( - os.environ.get("TWELVE_DATA_API_KEY") - or config.twelve_data_api_key - ) self._cache: Dict[str, pd.DataFrame] = {} self._cache_timestamps: Dict[str, float] = {} - self._session = requests.Session() - self._session.headers.update({ - "User-Agent": "Mozilla/5.0 (compatible; ForexIntel/1.0)", - }) - if self._api_key: - self._session.headers["Authorization"] = f"apikey {self._api_key}" - - @staticmethod - def _to_td_symbol(symbol: str) -> str: - """Convert internal symbol (EURUSD=X) to Twelve Data format (EUR/USD).""" - clean = symbol.replace("=X", "") - return f"{clean[:3]}/{clean[3:]}" def _cache_ttl(self, timeframe: Timeframe) -> int: """Cache TTL for a given timeframe — falls back to refresh_interval.""" @@ -157,7 +92,7 @@ class ForexDataProvider: if cached is not None and not cached.empty: return cached - df = self._fetch_twelve_data(symbol, timeframe, lookback_days) + df = self._fetch_yfinance(symbol, timeframe, lookback_days) df = self._validate_and_clean(df, symbol, timeframe) self._cache[cache_key] = df @@ -172,11 +107,6 @@ class ForexDataProvider: ) -> None: """ Batch-fetch all symbols for each timeframe in a single API call per TF. - - Twelve Data accepts comma-separated symbols, so 7 pairs × 3 timeframes - = 3 requests instead of 21 — well within the free-tier 8 req/min limit. - Results are stored in the local cache; subsequent fetch_ohlcv calls - for the same symbol/TF will be served from cache. """ if lookback_days is None: lookback_days = self.config.data_lookback_days @@ -191,11 +121,10 @@ class ForexDataProvider: for s in symbols ) if cached: - logger.debug(f"Skipping prefetch for {tf.value} — all {len(symbols)} symbols cached (TTL {ttl}s)") + 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 except Exception as e: logger.warning(f"Batch prefetch failed for {tf.value}: {e}") @@ -246,7 +175,6 @@ class ForexDataProvider: """Return display name and pip value for a pair.""" display = SYMBOL_DISPLAY.get(symbol, symbol) is_jpy = "JPY" in symbol - # Commodity-specific pip values take priority, then JPY pairs, then standard forex pip_value = PIP_VALUES.get(symbol, 0.01 if is_jpy else 0.0001) return { "symbol": symbol, @@ -257,27 +185,32 @@ class ForexDataProvider: # ── 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. - """ + def _fetch_yfinance(self, symbol: str, timeframe: Timeframe, lookback_days: int) -> pd.DataFrame: 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 + raise DataUnavailableError("yfinance is not installed.") - # 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"): + yf_symbol = _YF_FALLBACK_MAP.get(symbol, symbol) + + # Determine actual interval and resampling needs + actual_timeframe = timeframe + resample_needed = False + if timeframe == Timeframe.H4: + actual_timeframe = Timeframe.H1 + resample_needed = True + + yf_interval = _YF_INTERVALS.get(actual_timeframe) + if yf_interval is None: + raise DataUnavailableError(f"Unsupported timeframe {timeframe.value} for yfinance") + + # Period logic based on Yahoo limitations + if yf_interval in ("1m", "2m", "5m"): + period = "7d" if lookback_days < 7 else "60d" + elif yf_interval in ("15m", "30m", "1h", "60m"): period = "60d" - else: + elif yf_interval in ("1d", "1wk"): period = "2y" + else: + period = f"{lookback_days}d" try: df = yf.download( @@ -285,272 +218,122 @@ class ForexDataProvider: progress=False, auto_adjust=False, threads=False, ) if df is None or df.empty: - return None + raise DataUnavailableError(f"No data returned for {symbol}") - # 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}" - ) + + if resample_needed and timeframe == Timeframe.H4: + # Resample 1h data into 4h blocks + df = df.resample('4h').agg({ + 'Open': 'first', + 'High': 'max', + 'Low': 'min', + 'Close': 'last', + 'Volume': 'sum' + }).dropna() + + logger.info(f"[yf live] {yf_symbol} {timeframe.value}: {len(df)} bars, 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 + raise DataUnavailableError(f"yfinance fetch failed for {symbol} {timeframe.value}: {e}") - def _throttle(self, credits: int = 1) -> None: - """ - Block until `credits` per-minute API credits are available. - 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. - """ - global _rate_daily_credits, _rate_daily_reset - - with _rate_lock: - now = time.time() - - # 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 - - # 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 - ) -> None: - """ - Single API call fetching all symbols for one timeframe. - Twelve Data returns a dict keyed by symbol when multiple are requested. - Parsed DataFrames are validated and stored in the cache. - """ - if not self._api_key: + def _batch_fetch_timeframe(self, symbols: List[str], timeframe: Timeframe, lookback_days: int) -> None: + if not _YF_AVAILABLE: return - td_symbols = [self._to_td_symbol(s) for s in symbols] - interval = self.TD_INTERVALS[timeframe] - outputsize = min( - int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, 5000 - ) - outputsize = max(outputsize, 300) + actual_timeframe = timeframe + resample_needed = False + if timeframe == Timeframe.H4: + actual_timeframe = Timeframe.H1 + resample_needed = True - # 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)) + yf_interval = _YF_INTERVALS.get(actual_timeframe) + if not yf_interval: + return - resp = self._session.get( - f"{self.TD_BASE_URL}/time_series", - params={ - "symbol": ",".join(td_symbols), - "interval": interval, - "outputsize": outputsize, - "timezone": "UTC", - "format": "json", - }, - timeout=30, - ) - resp.raise_for_status() - data = resp.json() + yf_symbols = [_YF_FALLBACK_MAP.get(s, s) for s in symbols] - # If the whole batch request returned a top-level error, raise it - if isinstance(data, dict) and data.get("status") == "error": - raise DataIntegrityError( - f"Batch API error ({timeframe.value}): {data.get('message', 'unknown')}" - ) - - # Batch response: {"EUR/USD": {...}, "GBP/USD": {...}, ...} - now = time.time() - for symbol, td_sym in zip(symbols, td_symbols): - try: - sym_data = data.get(td_sym, {}) - if sym_data.get("status") == "error": - logger.warning( - f"Batch fetch error for {td_sym}: {sym_data.get('message')}" - ) - continue - - values = sym_data.get("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: - 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 = [ - { - "Datetime": pd.Timestamp(v["datetime"], tz="UTC"), - "Open": float(v["open"]) if v.get("open") is not None else None, - "High": float(v["high"]) if v.get("high") is not None else None, - "Low": float(v["low"]) if v.get("low") is not None else None, - "Close": float(v["close"]) if v.get("close") is not None else None, - "Volume": float(v["volume"]) if v.get("volume") is not None else 0.0, - } - for v in reversed(values) - if v.get("datetime") - ] - df = pd.DataFrame(records).set_index("Datetime") - df.index.name = "Datetime" - df = self._validate_and_clean(df, symbol, timeframe) - - cache_key = f"{symbol}_{timeframe.value}" - self._cache[cache_key] = df - self._cache_timestamps[cache_key] = now - logger.info(f"[live] {td_sym} {interval}: {len(df)} bars, last close={df['Close'].iloc[-1]:.5f}") - - except Exception as e: - logger.warning(f"Failed to parse batch data for {td_sym}: {e}") - - def _fetch_twelve_data( - self, symbol: str, timeframe: Timeframe, lookback_days: int - ) -> pd.DataFrame: - """ - Fetch OHLCV from Twelve Data REST API. - 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: - # 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( - "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] - outputsize = min( - int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, - 5000, - ) - outputsize = max(outputsize, 300) - - # Raises DataUnavailableError on daily limit; blocks on per-minute limit. - self._throttle(credits=1) + if yf_interval in ("1m", "2m", "5m"): + period = "7d" if lookback_days < 7 else "60d" + elif yf_interval in ("15m", "30m", "1h", "60m"): + period = "60d" + elif yf_interval in ("1d", "1wk"): + period = "2y" + else: + period = f"{lookback_days}d" try: - resp = self._session.get( - f"{self.TD_BASE_URL}/time_series", - params={ - "symbol": td_symbol, - "interval": interval, - "outputsize": outputsize, - "timezone": "UTC", - "format": "json", - }, - timeout=15, + df_multi = yf.download( + " ".join(yf_symbols), interval=yf_interval, period=period, + progress=False, auto_adjust=False, threads=True, group_by="ticker" ) - resp.raise_for_status() - data = resp.json() + + if df_multi is None or df_multi.empty: + logger.warning(f"Batch fetch empty for {timeframe.value}") + return - if data.get("status") == "error": - raise DataUnavailableError( - f"Twelve Data API error for {td_symbol}: {data.get('message', 'unknown')}" - ) + if df_multi.index.tz is None: + df_multi.index = df_multi.index.tz_localize("UTC") + else: + df_multi.index = df_multi.index.tz_convert("UTC") + df_multi.index.name = "Datetime" - values = data.get("values") - if not values: - yf_df = self._fetch_yfinance_fallback(symbol, timeframe) - if yf_df is not None: - return yf_df - raise DataUnavailableError( - f"No data returned for {td_symbol} — symbol may require a paid plan." - ) + now = time.time() + for symbol, yf_sym in zip(symbols, yf_symbols): + try: + if len(yf_symbols) == 1: + df_sym = df_multi.copy() + if isinstance(df_sym.columns, pd.MultiIndex): + df_sym.columns = df_sym.columns.get_level_values(0) + else: + if yf_sym in df_multi.columns.get_level_values(0): + df_sym = df_multi[yf_sym].copy() + else: + # Might be single level columns if download only returned 1 valid ticker + if yf_sym == yf_symbols[0] and not isinstance(df_multi.columns, pd.MultiIndex): + df_sym = df_multi.copy() + else: + continue - records = [ - { - "Datetime": pd.Timestamp(v["datetime"], tz="UTC"), - "Open": float(v["open"]) if v.get("open") is not None else None, - "High": float(v["high"]) if v.get("high") is not None else None, - "Low": float(v["low"]) if v.get("low") is not None else None, - "Close": float(v["close"]) if v.get("close") is not None else None, - "Volume": float(v["volume"]) if v.get("volume") is not None else 0.0, - } - for v in reversed(values) - if v.get("datetime") - ] + # Ignore if column names are not Open/High/Low/Close + if 'Close' not in df_sym.columns: + continue - df = pd.DataFrame(records).set_index("Datetime") - df.index.name = "Datetime" - logger.info(f"[live] {td_symbol} {interval}: {len(df)} bars, last close={df['Close'].iloc[-1]:.5f}") - return df + df_sym = df_sym[["Open", "High", "Low", "Close", "Volume"]].copy() + + if resample_needed and timeframe == Timeframe.H4: + df_sym = df_sym.resample('4h').agg({ + 'Open': 'first', + 'High': 'max', + 'Low': 'min', + 'Close': 'last', + 'Volume': 'sum' + }).dropna() + + df_sym = self._validate_and_clean(df_sym, symbol, timeframe) + cache_key = f"{symbol}_{timeframe.value}" + self._cache[cache_key] = df_sym + self._cache_timestamps[cache_key] = now + logger.info(f"[yf batch] {yf_sym} {timeframe.value}: {len(df_sym)} bars, last close={df_sym['Close'].iloc[-1]:.5f}") + except Exception as e: + logger.warning(f"Failed to process batch data for {symbol}: {e}") - except (DataUnavailableError, DataIntegrityError): - raise except Exception as e: - raise DataUnavailableError( - f"Failed to fetch {td_symbol}: {e}" - ) from e + logger.warning(f"Batch fetch failed for {timeframe.value}: {e}") def _validate_and_clean( self, df: pd.DataFrame, symbol: str, timeframe: Timeframe ) -> pd.DataFrame: """ Validate and clean OHLCV data. - - Steps: drop all-NaN rows → ffill small gaps → remove future timestamps - → fix OHLC inconsistencies → remove >10-sigma outliers → sort & deduplicate. """ if df is None or df.empty: raise DataIntegrityError(f"Empty dataset for {symbol}") @@ -564,7 +347,7 @@ class ForexDataProvider: now = pd.Timestamp.utcnow() future_mask = df.index > now + pd.Timedelta(minutes=5) if future_mask.any(): - logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps") + logger.debug(f"{symbol}: Dropping {future_mask.sum()} future timestamps") df = df[~future_mask] invalid_ohlc = ( @@ -575,7 +358,7 @@ class ForexDataProvider: (df["Low"] > df["Close"]) ) if invalid_ohlc.any(): - logger.warning(f"{symbol}: Fixing {invalid_ohlc.sum()} OHLC-inconsistent bars") + logger.debug(f"{symbol}: Fixing {invalid_ohlc.sum()} OHLC-inconsistent bars") for idx in df[invalid_ohlc].index: vals = [df.loc[idx, "Open"], df.loc[idx, "Close"]] df.loc[idx, "High"] = max(vals + [df.loc[idx, "High"]]) @@ -596,7 +379,7 @@ class ForexDataProvider: ) + 10 if len(df) < min_required: - logger.warning( + logger.debug( f"{symbol} {timeframe.value}: Only {len(df)} bars " f"(need {min_required}). Some indicators will be partial." ) @@ -606,6 +389,6 @@ class ForexDataProvider: cleaned = original_len - len(df) if cleaned > 0: - logger.info(f"{symbol}: Cleaned {cleaned}/{original_len} bars") + logger.debug(f"{symbol}: Cleaned {cleaned}/{original_len} bars") return df diff --git a/signal_aggregator.py b/signal_aggregator.py index 7b07e3c..fff4fd5 100644 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -17,6 +17,7 @@ import logging from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Dict, List, Optional +import concurrent.futures import pandas as pd @@ -124,13 +125,18 @@ class SignalAggregator: ) self.data_provider.prefetch_batch(self.config.watch_pairs, all_timeframes) - for symbol in self.config.watch_pairs: - try: - report = self.analyze_pair(symbol, context) - reports.append(report) - except Exception as e: - logger.error(f"Failed to analyze {symbol}: {e}") - continue + with concurrent.futures.ThreadPoolExecutor(max_workers=min(32, len(self.config.watch_pairs) or 1)) as executor: + future_to_symbol = { + executor.submit(self.analyze_pair, symbol, context): symbol + for symbol in self.config.watch_pairs + } + for future in concurrent.futures.as_completed(future_to_symbol): + symbol = future_to_symbol[future] + try: + report = future.result() + reports.append(report) + except Exception as e: + logger.error(f"Failed to analyze {symbol}: {e}") # Sort by absolute score (strongest signals first) reports.sort(key=lambda r: abs(r.final_score), reverse=True)