Harwinder
f5d6d35063
- New _PER_SYMBOL_TFS map in signal_aggregator routes gold/silver to M15+H1+H4 confirmation; forex pairs untouched. - H4 for XAU/XAG is resampled from yfinance H1 bars (yfinance has no native 4h candle). New _resample_h1_to_h4 helper. - Batch prefetch unions per-symbol TFs so future per-pair overrides work without extra plumbing. XAU/XAG still excluded from Twelve Data batch — H4 resample triggers a single yfinance H1 fetch. - Telegram SCAN_TIMEFRAMES restored to [M15, H1] so /scan emits both M15-primary and H1-primary cards. Shared cache means no extra credits. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
672 lines
25 KiB
Python
672 lines
25 KiB
Python
"""
|
||
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.
|
||
"""
|
||
|
||
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
|
||
_YF_AVAILABLE = True
|
||
except ImportError:
|
||
yf = None
|
||
_YF_AVAILABLE = False
|
||
|
||
# Insecure HTTPS session for yfinance — disables SSL verification.
|
||
# Needed on servers behind a TLS-inspecting proxy that re-signs certs with
|
||
# an untrusted root. yfinance requires a curl_cffi session, not requests.
|
||
# Security trade-off: man-in-the-middle attacks on yfinance traffic are
|
||
# possible. Acceptable here because the data is public market quotes.
|
||
try:
|
||
from curl_cffi import requests as _curl_cffi_requests
|
||
_YF_SESSION = _curl_cffi_requests.Session(impersonate="chrome", verify=False)
|
||
except ImportError:
|
||
_YF_SESSION = None
|
||
|
||
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).
|
||
_YF_FALLBACK_MAP = {
|
||
"XAUUSD=X": "GC=F",
|
||
"XAGUSD=X": "SI=F",
|
||
}
|
||
|
||
_YF_INTERVALS = {
|
||
Timeframe.M1: "1m",
|
||
Timeframe.M5: "5m",
|
||
Timeframe.M15: "15m",
|
||
Timeframe.M30: "30m",
|
||
Timeframe.H1: "1h",
|
||
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)."""
|
||
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.
|
||
"""
|
||
|
||
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."""
|
||
return self.config.cache_ttl_per_tf.get(timeframe, self.config.refresh_interval)
|
||
|
||
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()
|
||
ttl = self._cache_ttl(timeframe)
|
||
|
||
if (
|
||
cache_key in self._cache_timestamps
|
||
and (now - self._cache_timestamps[cache_key]) < ttl
|
||
and cache_key in self._cache
|
||
):
|
||
cached = self._cache.get(cache_key)
|
||
if cached is not None and not cached.empty:
|
||
return cached
|
||
|
||
# Symbols not on Twelve Data (gold, silver) always use yfinance.
|
||
# D1 is skipped (yfinance daily for futures is unreliable).
|
||
# H4 is resampled from H1 since yfinance has no native 4h candle.
|
||
if symbol in _YF_FALLBACK_MAP:
|
||
if timeframe == Timeframe.D1:
|
||
raise DataUnavailableError(
|
||
f"Daily chart skipped for {symbol} — gold/silver use M15/H1/H4 only."
|
||
)
|
||
if timeframe == Timeframe.H4:
|
||
h1_df = self._fetch_yfinance_fallback(symbol, Timeframe.H1)
|
||
if h1_df is None:
|
||
raise DataUnavailableError(
|
||
f"Yahoo Finance unavailable for {symbol} (H4 resample source H1)."
|
||
)
|
||
df = self._resample_h1_to_h4(h1_df)
|
||
df = self._validate_and_clean(df, symbol, timeframe)
|
||
self._cache[cache_key] = df
|
||
self._cache_timestamps[cache_key] = now
|
||
return df
|
||
yf_df = self._fetch_yfinance_fallback(symbol, timeframe)
|
||
if yf_df is None:
|
||
raise DataUnavailableError(
|
||
f"Yahoo Finance unavailable for {symbol} ({timeframe.value})."
|
||
)
|
||
df = self._validate_and_clean(yf_df, symbol, timeframe)
|
||
self._cache[cache_key] = df
|
||
self._cache_timestamps[cache_key] = now
|
||
return df
|
||
|
||
df = self._fetch_twelve_data(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:
|
||
"""
|
||
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
|
||
|
||
# XAU/XAG aren't on Twelve Data — exclude them from the batch call,
|
||
# they'll be fetched individually via yfinance on demand.
|
||
td_symbols = [s for s in symbols if s not in _YF_FALLBACK_MAP]
|
||
if not td_symbols:
|
||
return
|
||
|
||
now = time.time()
|
||
for tf in timeframes:
|
||
ttl = self._cache_ttl(tf)
|
||
cached = all(
|
||
f"{s}_{tf.value}" in self._cache_timestamps
|
||
and (now - self._cache_timestamps[f"{s}_{tf.value}"]) < ttl
|
||
and not self._cache.get(f"{s}_{tf.value}", pd.DataFrame()).empty
|
||
for s in td_symbols
|
||
)
|
||
if cached:
|
||
logger.debug(f"Skipping prefetch for {tf.value} — all {len(td_symbols)} symbols cached (TTL {ttl}s)")
|
||
continue
|
||
try:
|
||
self._batch_fetch_timeframe(td_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}")
|
||
|
||
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 = {}
|
||
for symbol in symbols:
|
||
try:
|
||
df = self.fetch_ohlcv(symbol, timeframe, lookback_days)
|
||
results[symbol] = df
|
||
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)
|
||
|
||
@staticmethod
|
||
def _resample_h1_to_h4(h1_df: pd.DataFrame) -> pd.DataFrame:
|
||
"""Resample 1-hour OHLCV to 4-hour bars (yfinance has no native 4h)."""
|
||
return h1_df.resample("4h", label="right", closed="right").agg({
|
||
"Open": "first",
|
||
"High": "max",
|
||
"Low": "min",
|
||
"Close": "last",
|
||
"Volume": "sum",
|
||
}).dropna()
|
||
|
||
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
|
||
# 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,
|
||
"display": display,
|
||
"pip_value": pip_value,
|
||
"is_jpy_pair": is_jpy,
|
||
}
|
||
|
||
# ── 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.
|
||
"""
|
||
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
|
||
|
||
# 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"):
|
||
period = "60d"
|
||
else:
|
||
period = "2y"
|
||
|
||
try:
|
||
yf_kwargs = dict(
|
||
interval=yf_interval, period=period,
|
||
progress=False, auto_adjust=False, threads=False,
|
||
)
|
||
if _YF_SESSION is not None:
|
||
yf_kwargs["session"] = _YF_SESSION
|
||
df = yf.download(yf_symbol, **yf_kwargs)
|
||
if df is None or df.empty:
|
||
return None
|
||
|
||
# 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}"
|
||
)
|
||
return df
|
||
except Exception as e:
|
||
logger.warning(f"yfinance fallback failed for {symbol} {timeframe.value}: {e}")
|
||
return None
|
||
|
||
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:
|
||
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)
|
||
|
||
# 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",
|
||
params={
|
||
"symbol": ",".join(td_symbols),
|
||
"interval": interval,
|
||
"outputsize": outputsize,
|
||
"timezone": "UTC",
|
||
"format": "json",
|
||
},
|
||
timeout=30,
|
||
)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
|
||
# 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)
|
||
|
||
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,
|
||
)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
|
||
if data.get("status") == "error":
|
||
raise DataUnavailableError(
|
||
f"Twelve Data API error for {td_symbol}: {data.get('message', 'unknown')}"
|
||
)
|
||
|
||
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."
|
||
)
|
||
|
||
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"
|
||
logger.info(f"[live] {td_symbol} {interval}: {len(df)} bars, last close={df['Close'].iloc[-1]:.5f}")
|
||
return df
|
||
|
||
except (DataUnavailableError, DataIntegrityError):
|
||
raise
|
||
except Exception as e:
|
||
raise DataUnavailableError(
|
||
f"Failed to fetch {td_symbol}: {e}"
|
||
) from 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}")
|
||
|
||
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.debug(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
|