From 43cc4bdd42b1cc254211da42151349d75a0bc257 Mon Sep 17 00:00:00 2001 From: Harwinder Date: Mon, 4 May 2026 20:12:41 -0400 Subject: [PATCH] Switch data source to Finnhub API; remove synthetic data fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds FinnhubDataProvider as a drop-in replacement for ForexDataProvider. Rate limit or API errors raise DataUnavailableError immediately — the Telegram bot shows a clean per-TF error card rather than fake prices. - data_provider_finnhub.py: Finnhub client, OANDA symbol mapping, 4h resampled from 60-min candles, shared module-level rate limiter (60 calls/min free tier), full OHLCV validation copied from original - signal_aggregator.py: import FinnhubDataProvider alias ForexDataProvider - config.py: add finnhub_api_key field (FINNHUB_API_KEY env var overrides) - telegram_bot.py: remove synthetic-data warning and _SUPPORTED_PAIRS; unsupported pairs now show the engine's "data unavailable" card - requirements.txt: add finnhub-python>=1.0 Set FINNHUB_API_KEY env var before running. main branch is untouched. Co-Authored-By: Claude Sonnet 4.6 --- config.py | 3 + data_provider_finnhub.py | 386 +++++++++++++++++++++++++++++++++++++++ requirements.txt | 1 + signal_aggregator.py | 2 +- telegram_bot.py | 19 +- 5 files changed, 397 insertions(+), 14 deletions(-) create mode 100644 data_provider_finnhub.py diff --git a/config.py b/config.py index f638681..d98919d 100644 --- a/config.py +++ b/config.py @@ -196,6 +196,9 @@ class PlatformConfig: # Twelve Data API key (overridden by TWELVE_DATA_API_KEY env var) twelve_data_api_key: str = "f563b22ffdfd4924955decbe1a8c8701" + # Finnhub API key (overridden by FINNHUB_API_KEY env var) + finnhub_api_key: str = "" + # Dashboard refresh interval (seconds) refresh_interval: int = 300 # 5 minutes diff --git a/data_provider_finnhub.py b/data_provider_finnhub.py new file mode 100644 index 0000000..f573e76 --- /dev/null +++ b/data_provider_finnhub.py @@ -0,0 +1,386 @@ +""" +Forex Signal Intelligence Platform - Finnhub Data Provider +=========================================================== +Drop-in replacement for ForexDataProvider using Finnhub instead of Twelve Data. +Raises DataUnavailableError on rate limit or API errors — no synthetic fallback. + +Setup: + set FINNHUB_API_KEY=your_key_here (Windows CMD) + export FINNHUB_API_KEY=your_key (Linux/Mac) + +Free tier limits: 60 API calls/minute. +Symbol broker: OANDA (e.g. OANDA:EUR_USD). +""" + +import logging +import os +import threading +import time +from collections import deque +from typing import Dict, List, Optional + +import pandas as pd +import finnhub + +from config import PlatformConfig, Timeframe, SYMBOL_DISPLAY, PIP_VALUES + +logger = logging.getLogger(__name__) + + +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).""" + pass + + +# Module-level rate limiter shared across all provider instances and threads. +_rate_lock = threading.Lock() +_rate_request_times: deque = deque() + +RATE_LIMIT = 60 # Finnhub free tier: 60 calls/minute +RATE_WINDOW = 60.0 + + +class FinnhubDataProvider: + """ + Forex data provider backed by the Finnhub API (OANDA feed). + + Free tier: 60 calls/minute, no hard daily cap. + When the rate limit or any API error occurs, DataUnavailableError is raised + immediately — there is no synthetic/simulated data fallback. + + Supported resolutions: 1, 5, 15, 30, 60, D, W. + H4 bars are synthesised client-side by resampling 60-minute candles. + """ + + # Finnhub resolution strings (H4 uses 60-min candles, then resampled) + FH_RESOLUTIONS = { + Timeframe.M1: "1", + Timeframe.M5: "5", + Timeframe.M15: "15", + Timeframe.M30: "30", + Timeframe.H1: "60", + Timeframe.H4: "60", + Timeframe.D1: "D", + Timeframe.W1: "W", + } + + def __init__(self, config: PlatformConfig): + self.config = config + api_key = ( + os.environ.get("FINNHUB_API_KEY") + or getattr(config, "finnhub_api_key", "") + ) + if not api_key: + raise DataUnavailableError( + "No Finnhub API key — set the FINNHUB_API_KEY environment variable." + ) + self._client = finnhub.Client(api_key=api_key) + self._cache: Dict[str, pd.DataFrame] = {} + self._cache_timestamps: Dict[str, float] = {} + + @staticmethod + def _to_finnhub_symbol(symbol: str) -> str: + """Convert EURUSD=X → OANDA:EUR_USD.""" + clean = symbol.replace("=X", "") + return f"OANDA:{clean[:3]}_{clean[3:]}" + + # ── Public interface (matches ForexDataProvider) ─────────────────── + + def fetch_ohlcv( + self, + symbol: str, + timeframe: Timeframe, + lookback_days: Optional[int] = None, + ) -> pd.DataFrame: + """ + Fetch OHLCV data for a Forex pair with full validation. + + Returns a DataFrame with columns: Open, High, Low, Close, Volume. + Index is a timezone-aware DatetimeIndex in UTC. + """ + if lookback_days is None: + lookback_days = self.config.data_lookback_days + + cache_key = f"{symbol}_{timeframe.value}" + now = time.time() + + if ( + cache_key in self._cache_timestamps + and (now - self._cache_timestamps[cache_key]) < self.config.refresh_interval + and cache_key in self._cache + ): + cached = self._cache.get(cache_key) + if cached is not None and not cached.empty: + return cached + + df = self._fetch_finnhub(symbol, timeframe, lookback_days) + df = self._validate_and_clean(df, symbol, timeframe) + + self._cache[cache_key] = df + self._cache_timestamps[cache_key] = now + return df + + def prefetch_batch( + self, + symbols: List[str], + timeframes: List[Timeframe], + lookback_days: Optional[int] = None, + ) -> None: + """ + Sequentially prefetch symbols × timeframes. + + Finnhub has no batch endpoint, so we fetch one symbol at a time. + The module-level rate limiter ensures we never exceed 60 calls/minute + even when multiple threads call this concurrently. + """ + if lookback_days is None: + lookback_days = self.config.data_lookback_days + + for tf in timeframes: + for symbol in symbols: + try: + self.fetch_ohlcv(symbol, tf, lookback_days) + except DataUnavailableError as e: + logger.warning(f"Prefetch skipped {symbol} {tf.value}: {e}") + except Exception as e: + logger.warning(f"Prefetch failed {symbol} {tf.value}: {e}") + + def fetch_multiple( + self, + symbols: List[str], + timeframe: Timeframe, + lookback_days: Optional[int] = None, + ) -> Dict[str, pd.DataFrame]: + """Fetch data for multiple symbols, returning dict of DataFrames.""" + results: Dict[str, pd.DataFrame] = {} + for symbol in symbols: + try: + results[symbol] = self.fetch_ohlcv(symbol, timeframe, lookback_days) + except Exception as e: + logger.warning(f"Failed to fetch {symbol}: {e}") + return results + + def get_latest_price(self, symbol: str) -> Optional[float]: + """Return the most recent close price for a symbol.""" + try: + df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1) + if df is not None and not df.empty: + return float(df["Close"].iloc[-1]) + except Exception: + pass + return None + + def get_spread_estimate(self, symbol: str) -> float: + """Estimate typical spread from recent M5 high-low ranges.""" + try: + df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1) + if df is not None and not df.empty: + return float((df["High"] - df["Low"]).median()) + except Exception: + pass + + fallback_spreads = { + "EURUSD=X": 0.00010, "GBPUSD=X": 0.00015, + "USDJPY=X": 0.010, "USDCHF=X": 0.00015, + "AUDUSD=X": 0.00015, "USDCAD=X": 0.00018, + "NZDUSD=X": 0.00020, + } + return fallback_spreads.get(symbol, 0.00020) + + def get_pair_info(self, symbol: str) -> Dict: + """Return display name and pip value for a pair.""" + display = SYMBOL_DISPLAY.get(symbol, symbol) + is_jpy = "JPY" in symbol + pip_value = PIP_VALUES.get(symbol, 0.01 if is_jpy else 0.0001) + return { + "symbol": symbol, + "display": display, + "pip_value": pip_value, + "is_jpy_pair": is_jpy, + } + + # ── Private Methods ──────────────────────────────────────────────── + + def _throttle(self) -> None: + """ + Block until a per-minute API call slot is available. + + Uses module-level state so all FinnhubDataProvider instances + (including those in parallel threads) share the same budget. + """ + with _rate_lock: + while True: + now = time.time() + while _rate_request_times and now - _rate_request_times[0] >= RATE_WINDOW: + _rate_request_times.popleft() + + if len(_rate_request_times) < RATE_LIMIT: + _rate_request_times.append(time.time()) + return + + wait = (_rate_request_times[0] + RATE_WINDOW) - now + 0.2 + logger.info( + f"Finnhub rate limit: {len(_rate_request_times)}/{RATE_LIMIT} " + f"calls used — waiting {wait:.1f}s…" + ) + _rate_lock.release() + time.sleep(wait) + _rate_lock.acquire() + + def _fetch_finnhub( + self, symbol: str, timeframe: Timeframe, lookback_days: int + ) -> pd.DataFrame: + """ + Fetch OHLCV from Finnhub REST API via the official Python client. + Raises DataUnavailableError on any API failure or rate limit. + """ + fh_symbol = self._to_finnhub_symbol(symbol) + resolution = self.FH_RESOLUTIONS[timeframe] + + to_ts = int(time.time()) + # For H4 resampling we need the full 1h candle history + from_ts = to_ts - (lookback_days * 86400) + + self._throttle() + + try: + data = self._client.forex_candles(fh_symbol, resolution, from_ts, to_ts) + except finnhub.FinnhubAPIException as e: + if e.status_code == 429: + raise DataUnavailableError( + "Finnhub API rate limit reached — too many requests. " + "Free tier allows 60 calls/minute. Try again shortly." + ) from e + raise DataUnavailableError( + f"Finnhub API error for {fh_symbol} ({e.status_code}): {e}" + ) from e + except Exception as e: + raise DataUnavailableError( + f"Failed to fetch {fh_symbol} from Finnhub: {e}" + ) from e + + status = data.get("s") if data else None + + if status == "no_data": + raise DataUnavailableError( + f"Finnhub returned no data for {fh_symbol} — " + "symbol may not be supported on your plan, or the market is closed." + ) + + if status != "ok": + raise DataUnavailableError( + f"Finnhub returned unexpected status '{status}' for {fh_symbol}." + ) + + timestamps = data.get("t", []) + if not timestamps: + raise DataUnavailableError( + f"Finnhub response for {fh_symbol} contained no candles." + ) + + records = [ + { + "Datetime": pd.Timestamp(ts, unit="s", tz="UTC"), + "Open": float(o), + "High": float(h), + "Low": float(l), + "Close": float(c), + "Volume": float(v), + } + for ts, o, h, l, c, v in zip( + data["t"], data["o"], data["h"], data["l"], data["c"], data["v"] + ) + ] + + df = pd.DataFrame(records).set_index("Datetime") + df.index.name = "Datetime" + + if timeframe == Timeframe.H4: + df = self._resample_to_4h(df) + + logger.debug(f"Fetched {len(df)} bars for {fh_symbol} @ {resolution}") + return df + + @staticmethod + def _resample_to_4h(df: pd.DataFrame) -> pd.DataFrame: + """Resample 1-hour OHLCV candles into 4-hour bars.""" + resampled = df.resample("4h").agg({ + "Open": "first", + "High": "max", + "Low": "min", + "Close": "last", + "Volume": "sum", + }) + return resampled.dropna(subset=["Close"]) + + 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}") + + original_len = len(df) + + df = df.dropna(subset=["Open", "High", "Low", "Close"], how="all") + df = df.ffill(limit=3) + df = df.dropna(subset=["Close"]) + + 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") + df = df[~future_mask] + + invalid_ohlc = ( + (df["High"] < df["Low"]) | + (df["High"] < df["Open"]) | + (df["High"] < df["Close"]) | + (df["Low"] > df["Open"]) | + (df["Low"] > df["Close"]) + ) + if invalid_ohlc.any(): + logger.warning(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"]]) + df.loc[idx, "Low"] = min(vals + [df.loc[idx, "Low"]]) + + returns = df["Close"].pct_change() + std_ret = returns.std() + if std_ret > 0: + outlier_mask = (returns - returns.mean()).abs() > 10 * std_ret + if outlier_mask.any(): + logger.warning(f"{symbol}: Removing {outlier_mask.sum()} extreme outlier bars") + df = df[~outlier_mask] + + min_required = max( + self.config.technical.ema_trend, + self.config.technical.ichimoku_senkou_b, + 200, + ) + 10 + + if len(df) < min_required: + logger.warning( + f"{symbol} {timeframe.value}: Only {len(df)} bars " + f"(need {min_required}). Some indicators will be partial." + ) + + df = df.sort_index() + df = df[~df.index.duplicated(keep="last")] + + cleaned = original_len - len(df) + if cleaned > 0: + logger.info(f"{symbol}: Cleaned {cleaned}/{original_len} bars") + + return df diff --git a/requirements.txt b/requirements.txt index 615386f..2a09f29 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ numpy>=1.24 pandas>=2.0 requests>=2.28 python-telegram-bot[job-queue]>=20.0 +finnhub-python>=1.0 diff --git a/signal_aggregator.py b/signal_aggregator.py index 38122bb..861b9fa 100644 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -24,7 +24,7 @@ from config import ( PlatformConfig, SignalStrength, Timeframe, SYMBOL_DISPLAY, ) -from data_provider import ForexDataProvider +from data_provider_finnhub import FinnhubDataProvider as ForexDataProvider from engine import TechnicalEngine, TechnicalAnalysis from news_engine import NewsEngine, FundamentalContext diff --git a/telegram_bot.py b/telegram_bot.py index 63753cd..ac7a648 100644 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -30,7 +30,7 @@ from telegram.constants import ParseMode from telegram.ext import Application, CommandHandler, ContextTypes from config import ( - MAJOR_PAIRS, CROSS_PAIRS, SYMBOL_DISPLAY, + MAJOR_PAIRS, SYMBOL_DISPLAY, PlatformConfig, SignalStrength, Timeframe, ) from signal_aggregator import SignalAggregator, SignalReport @@ -54,8 +54,7 @@ _TF_CONFIRMATIONS = { "D1": [Timeframe.H4, Timeframe.W1], } -_SUPPORTED_PAIRS = set(MAJOR_PAIRS + CROSS_PAIRS) -_GRADE_RANK = {"A+": 0, "A": 1, "B": 2, "C": 3, "D": 4, "F": 5} +_GRADE_RANK = {"A+": 0, "A": 1, "B": 2, "C": 3, "D": 4, "F": 5} _EST = ZoneInfo("America/New_York") # ── Formatting helpers ────────────────────────────────────────────── @@ -114,7 +113,6 @@ def _build_combined_signal( results_by_tf: Dict[str, List[SignalReport]], display: str, now_str: str, - synthetic: bool = False, ) -> str: """ Single message showing all timeframes stacked. @@ -125,11 +123,7 @@ def _build_combined_signal( lines.append(f"🌐 {_h(display)} — Signal Analysis") lines.append(f"🕐 {now_str}") - if synthetic: - lines.append( - "⚠️ Note: Not on Twelve Data free tier — " - "prices are simulated data." - ) + # Finnhub branch: no synthetic data — real prices or error shown per-TF. for tf in SCAN_TIMEFRAMES: reports = results_by_tf.get(tf, []) @@ -360,9 +354,8 @@ class ForexTelegramBot: raw = args[0].upper().replace("/", "").replace("_", "") if not raw.endswith("=X"): raw += "=X" - display = SYMBOL_DISPLAY.get(raw, raw.replace("=X", "")) - tfs_str = " · ".join(SCAN_TIMEFRAMES) - synthetic = raw not in _SUPPORTED_PAIRS + display = SYMBOL_DISPLAY.get(raw, raw.replace("=X", "")) + tfs_str = " · ".join(SCAN_TIMEFRAMES) msg = await update.message.reply_text( f"🔄 Analyzing {_h(display)} across {_h(tfs_str)}…" @@ -376,7 +369,7 @@ class ForexTelegramBot: await msg.edit_text(f"❌ No data returned for {_h(display)}") return - text = _build_combined_signal(results_by_tf, display, now_str, synthetic) + text = _build_combined_signal(results_by_tf, display, now_str) await msg.delete() await self._send_html(update, text)