Switch data source to Finnhub API; remove synthetic data fallback
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 <noreply@anthropic.com>
This commit is contained in:
Harwinder
@@ -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
|
||||
Reference in New Issue
Block a user