BluefxPraise/The_Oracle
0
1"""Brute-force EMA / RSI / threshold parameter sweeps."""2 3import logging4from typing import Any, Dict, List5 6import numpy as np7import pandas as pd8 9import memory10from core.risk_engine import compute_sl_tp, atr_value11 12log = logging.getLogger("oracle.tuner")13 14 15def _ema(s: pd.Series, n: int) -> pd.Series:16 return s.ewm(span=n, adjust=False).mean()17 18 19def _rsi(s: pd.Series, n: int = 14) -> pd.Series:20 d = s.diff()21 g = d.where(d > 0, 0.0).rolling(n).mean()22 l = (-d.where(d < 0, 0.0)).rolling(n).mean()23 rs = g / l.replace(0, np.nan)24 return (100 - 100 / (1 + rs)).fillna(50)25 26 27def _simulate(pair: str, candles: List[Dict], ema_fast: int, ema_slow: int,28 rsi_n: int, threshold: float) -> Dict[str, Any]:29 if len(candles) < 80 or ema_fast >= ema_slow:30 return {"pnl": -1e9, "trades": 0, "winrate": 0}31 df = pd.DataFrame(candles)32 df["ef"] = _ema(df["c"], ema_fast)33 df["es"] = _ema(df["c"], ema_slow)34 df["rsi"] = _rsi(df["c"], rsi_n)35 36 trades = []37 open_t = None38 for i in range(60, len(candles) - 1):39 bar = candles[i + 1]40 if open_t:41 hi, lo = float(bar["h"]), float(bar["l"])42 t = open_t43 if t["dir"] == "BUY":44 if lo <= t["sl"]:45 t["pnl"] = (t["sl"] - t["entry"]) / t["entry"]; t["w"] = 046 trades.append(t); open_t = None; continue47 if hi >= t["tp"]:48 t["pnl"] = (t["tp"] - t["entry"]) / t["entry"]; t["w"] = 149 trades.append(t); open_t = None; continue50 else:51 if hi >= t["sl"]:52 t["pnl"] = (t["entry"] - t["sl"]) / t["entry"]; t["w"] = 053 trades.append(t); open_t = None; continue54 if lo <= t["tp"]:55 t["pnl"] = (t["entry"] - t["tp"]) / t["entry"]; t["w"] = 156 trades.append(t); open_t = None; continue57 if open_t is None:58 ef, es = df["ef"].iloc[i], df["es"].iloc[i]59 rsi = df["rsi"].iloc[i]60 close = float(candles[i]["c"])61 score = 0.062 direction = None63 if ef > es and rsi > 50:64 direction = "BUY"; score = (ef - es) / max(1e-9, close) * 1000 + (rsi - 50)65 elif ef < es and rsi < 50:66 direction = "SELL"; score = (es - ef) / max(1e-9, close) * 1000 + (50 - rsi)67 confidence = min(100.0, max(0.0, 50 + score))68 if direction and confidence >= threshold:69 atr = atr_value(candles[: i + 1], 14)70 sl, tp, _ = compute_sl_tp(direction, close, atr, pair)71 open_t = {"dir": direction, "entry": close, "sl": sl, "tp": tp}72 73 wins = sum(t["w"] for t in trades)74 pnl = sum(t["pnl"] for t in trades)75 return {76 "pnl": float(pnl), "trades": len(trades),77 "winrate": 100.0 * wins / max(1, len(trades)),78 }79 80 81def sweep(pair: str, candles: List[Dict[str, Any]]) -> Dict[str, Any]:82 ema_fasts = [8, 12, 16]83 ema_slows = [20, 26, 34]84 rsi_ns = [9, 14, 21]85 thresholds = [60, 70, 80, 90]86 best = {"pnl": -1e9, "params": None, "metrics": None}87 grid = []88 for ef in ema_fasts:89 for es in ema_slows:90 if es <= ef:91 continue92 for rn in rsi_ns:93 for th in thresholds:94 m = _simulate(pair, candles, ef, es, rn, th)95 grid.append({"ef": ef, "es": es, "rsi": rn, "thr": th, **m})96 if m["pnl"] > best["pnl"] and m["trades"] >= 3:97 best = {"pnl": m["pnl"], "params": {"ef": ef, "es": es, "rsi": rn, "thr": th},98 "metrics": m}99 memory.log_calibration(100 kind="parameter_sweep",101 params={"pair": pair, "best": best["params"]},102 metrics={"best": best["metrics"], "grid_size": len(grid)},103 applied=False,104 )105 return {"best": best, "grid_size": len(grid)}106 