BluefxPraise/The_Oracle

1"""Market data + global rate-limit throttle."""2 3import logging4import threading5import time6from typing import Any, Dict, List, Optional7 8import pandas as pd9import yfinance as yf10 11import memory12from config import TWELVEDATA_API_KEY13 14log = logging.getLogger("oracle.data")15 16_THROTTLE_LOCK = threading.Lock()17_THROTTLE: Dict[str, float] = {}18_TWELVE_CALLS_TODAY: Dict[str, int] = {"date": "", "n": 0}19 20 21def throttle(key: str, seconds: float) -> bool:22    """Return True if action allowed (and reserve slot), False if throttled."""23    with _THROTTLE_LOCK:24        now = time.time()25        last = _THROTTLE.get(key, 0)26        if now - last < seconds:27            return False28        _THROTTLE[key] = now29        return True30 31 32def time_until(key: str, seconds: float) -> float:33    with _THROTTLE_LOCK:34        last = _THROTTLE.get(key, 0)35        delta = seconds - (time.time() - last)36        return max(0.0, delta)37 38 39# Cache TTL by timeframe (seconds)40_TTL = {"15m": 300, "1h": 1200, "4h": 3600}41 42 43def _yf_period_interval(timeframe: str) -> tuple:44    if timeframe == "15m":45        return ("15d", "15m")46    if timeframe == "1h":47        return ("60d", "60m")48    if timeframe == "4h":49        # yfinance has no native 4h; use 1h then resample50        return ("60d", "60m")51    return ("15d", timeframe)52 53 54def _to_candles(df: pd.DataFrame) -> List[Dict[str, Any]]:55    out = []56    for ts, row in df.iterrows():57        try:58            out.append({59                "t": int(pd.Timestamp(ts).timestamp()),60                "o": float(row["Open"]),61                "h": float(row["High"]),62                "l": float(row["Low"]),63                "c": float(row["Close"]),64                "v": float(row.get("Volume", 0) or 0),65            })66        except Exception:67            continue68    return out69 70 71def _resample_to_4h(candles: List[Dict[str, Any]]) -> List[Dict[str, Any]]:72    if not candles:73        return []74    df = pd.DataFrame(candles)75    df["dt"] = pd.to_datetime(df["t"], unit="s", utc=True)76    df = df.set_index("dt")77    rs = df.resample("4H").agg({"o": "first", "h": "max", "l": "min", "c": "last", "v": "sum"}).dropna()78    out = []79    for ts, row in rs.iterrows():80        out.append({81            "t": int(ts.timestamp()),82            "o": float(row["o"]), "h": float(row["h"]),83            "l": float(row["l"]), "c": float(row["c"]), "v": float(row["v"]),84        })85    return out86 87 88def _twelvedata_fetch(pair: str, timeframe: str, limit: int = 200) -> Optional[List[Dict[str, Any]]]:89    today = time.strftime("%Y-%m-%d", time.gmtime())90    with _THROTTLE_LOCK:91        if _TWELVE_CALLS_TODAY["date"] != today:92            _TWELVE_CALLS_TODAY["date"] = today93            _TWELVE_CALLS_TODAY["n"] = 094        if _TWELVE_CALLS_TODAY["n"] >= 800:95            log.warning("TwelveData daily limit reached")96            return None97        _TWELVE_CALLS_TODAY["n"] += 198 99    interval_map = {"15m": "15min", "1h": "1h", "4h": "4h"}100    sym = pair.replace("=X", "").replace("-USD", "/USD")101    if "/" not in sym and len(sym) == 6:102        sym = f"{sym[:3]}/{sym[3:]}"103    interval = interval_map.get(timeframe, timeframe)104    try:105        import requests106        r = requests.get(107            "https://api.twelvedata.com/time_series",108            params={109                "symbol": sym,110                "interval": interval,111                "outputsize": limit,112                "apikey": TWELVEDATA_API_KEY,113                "format": "JSON",114            },115            timeout=15,116        )117        j = r.json()118        if "values" not in j:119            log.warning("TwelveData no values for %s: %s", pair, j.get("message"))120            return None121        out = []122        for row in reversed(j["values"]):123            try:124                ts = int(pd.Timestamp(row["datetime"]).timestamp())125                out.append({126                    "t": ts,127                    "o": float(row["open"]),128                    "h": float(row["high"]),129                    "l": float(row["low"]),130                    "c": float(row["close"]),131                    "v": float(row.get("volume", 0) or 0),132                })133            except Exception:134                continue135        return out136    except Exception as e:137        log.warning("TwelveData error: %s", e)138        return None139 140 141def get_candles(pair: str, timeframe: str, limit: int = 200, force: bool = False) -> List[Dict[str, Any]]:142    ttl = _TTL.get(timeframe, 300)143    if not force:144        cached = memory.get_cached_candles(pair, timeframe, ttl)145        if cached:146            return cached[-limit:]147 148    key = f"yf:{pair}:{timeframe}"149    if not throttle(key, max(60.0, ttl / 4)):150        cached = memory.get_cached_candles(pair, timeframe, ttl * 4)151        if cached:152            return cached[-limit:]153 154    candles: List[Dict[str, Any]] = []155    try:156        period, interval = _yf_period_interval(timeframe)157        df = yf.download(158            pair, period=period, interval=interval,159            progress=False, threads=False, auto_adjust=False,160        )161        if df is not None and len(df) > 0:162            if isinstance(df.columns, pd.MultiIndex):163                df.columns = df.columns.get_level_values(0)164            candles = _to_candles(df)165            if timeframe == "4h":166                candles = _resample_to_4h(candles)167    except Exception as e:168        log.warning("yfinance error %s %s: %s", pair, timeframe, e)169 170    if not candles:171        td = _twelvedata_fetch(pair, timeframe, limit)172        if td:173            candles = td174 175    if candles:176        memory.cache_candles(pair, timeframe, candles)177    return candles[-limit:] if candles else []178 179 180def get_latest_price(pair: str) -> float:181    candles = get_candles(pair, "15m", limit=2)182    if candles:183        return float(candles[-1]["c"])184    return 0.0185