BluefxPraise/The_Oracle

1"""Historical replay backtester."""2 3import logging4from typing import Any, Dict, List, Optional5 6import pandas as pd7 8import memory9from config import ATR_BUFFERS, MIN_RR10from core import trend_engine, mean_reversion_engine, ml_classifier11from core.risk_engine import compute_sl_tp, atr_value12 13log = logging.getLogger("oracle.backtest")14 15 16def _evaluate_window(candles: List[Dict], weights: Dict[str, float], threshold: float):17    results = []18    for r in (19        trend_engine.evaluate(candles),20        mean_reversion_engine.evaluate(candles),21    ):22        if r:23            results.append(r)24    if len(results) < 2:25        return None26    buys = [r for r in results if r["direction"] == "BUY"]27    sells = [r for r in results if r["direction"] == "SELL"]28    if len(buys) >= 2:29        agreeing, direction = buys, "BUY"30    elif len(sells) >= 2:31        agreeing, direction = sells, "SELL"32    else:33        return None34    total_w = sum(weights.get(r["model"], 1.0) for r in agreeing) or 1.035    conf = sum(weights.get(r["model"], 1.0) * r["confidence"] for r in agreeing) / total_w36    if conf < threshold:37        return None38    return {"direction": direction, "confidence": conf}39 40 41def run(pair: str, candles: List[Dict[str, Any]], days: int,42        weights: Optional[Dict[str, float]] = None,43        threshold: float = 70.0) -> Dict[str, Any]:44    if not candles or len(candles) < 80:45        return {"trades": 0, "wins": 0, "losses": 0, "winrate": 0, "pnl": 0,46                "max_dd": 0, "sharpe": 0}47    weights = weights or {"trend": 2.0, "mr": 1.0, "ml": 1.5}48    trades = []49    open_trade = None50    equity = [0.0]51    for i in range(60, len(candles) - 1):52        window = candles[: i + 1]53        bar = candles[i + 1]54        if open_trade:55            hi, lo = float(bar["h"]), float(bar["l"])56            t = open_trade57            if t["direction"] == "BUY":58                if lo <= t["sl"]:59                    t["pnl"] = (t["sl"] - t["entry"]) / t["entry"]60                    t["outcome"] = "LOSS"61                    trades.append(t); open_trade = None62                    equity.append(equity[-1] + t["pnl"])63                    continue64                if hi >= t["tp"]:65                    t["pnl"] = (t["tp"] - t["entry"]) / t["entry"]66                    t["outcome"] = "WIN"67                    trades.append(t); open_trade = None68                    equity.append(equity[-1] + t["pnl"])69                    continue70            else:71                if hi >= t["sl"]:72                    t["pnl"] = (t["entry"] - t["sl"]) / t["entry"]73                    t["outcome"] = "LOSS"74                    trades.append(t); open_trade = None75                    equity.append(equity[-1] + t["pnl"])76                    continue77                if lo <= t["tp"]:78                    t["pnl"] = (t["entry"] - t["tp"]) / t["entry"]79                    t["outcome"] = "WIN"80                    trades.append(t); open_trade = None81                    equity.append(equity[-1] + t["pnl"])82                    continue83        if open_trade is None:84            sig = _evaluate_window(window, weights, threshold)85            if sig:86                entry = float(window[-1]["c"])87                atr = atr_value(window, 14)88                sl, tp, _ = compute_sl_tp(sig["direction"], entry, atr, pair)89                open_trade = {"direction": sig["direction"], "entry": entry, "sl": sl, "tp": tp,90                              "conf": sig["confidence"]}91    wins = sum(1 for t in trades if t["outcome"] == "WIN")92    losses = sum(1 for t in trades if t["outcome"] == "LOSS")93    pnl = sum(t["pnl"] for t in trades)94    eq = pd.Series(equity)95    rolling_max = eq.cummax()96    drawdowns = eq - rolling_max97    max_dd = float(drawdowns.min()) if len(drawdowns) else 0.098    rets = eq.diff().dropna()99    sharpe = float((rets.mean() / rets.std()) * (252 ** 0.5)) if len(rets) > 2 and rets.std() else 0.0100    metrics = {101        "trades": len(trades), "wins": wins, "losses": losses,102        "winrate": 100.0 * wins / max(1, wins + losses),103        "pnl": float(pnl), "max_dd": max_dd, "sharpe": sharpe,104    }105    memory.save_backtest(pair, days, "default", metrics,106                         {"weights": weights, "threshold": threshold})107    return metrics108