"""Walk-forward backtester — backtest.py Replays 6 months of 15m candles through the same Markov state classifier and signal detectors used in live trading. No lookahead bias: at bar i, only data[0:i] is visible. IMPORTANT DESIGN DECISIONS (v2): - Uses 15m candles (same timeframe as scorer.py SL/TP logic) - Uses 15m ATR for SL/TP (same multipliers as live card) - Tracks BOTH TP1 and TP2 win conditions separately: TP1 win = TP1 hit before SL (conservative, matches how most traders exit) TP2 win = TP2 hit before SL (ambitious, longer hold) - Signal detector: same 5 indicators as before, tuned for 15m bars Win conditions: TP1 win: price hits TP1 before SL (1.5× ATR target) TP2 win: price hits TP2 before SL (2.5× ATR target) For each simulated trade it records: - entry price, SL, TP1, TP2 - Markov state at entry - direction (long/short) - outcome_tp1 (win / loss / timeout) - outcome_tp2 (win / loss / timeout) - bars_to_outcome Aggregates: - TP1 win rate (primary — matches user's actual trading style) - TP2 win rate (secondary — for ambitious hold targets) - win rate by state (BULL/BEAR/RANGING) - average R:R realised - equity curve (cumulative P&L in R-multiples based on TP1) - per-trade list (last 50) Cache: 6h per symbol. """ from __future__ import annotations import math, time, statistics from markov import classify_state, ALL_STATES # ─── Cache ─────────────────────────────────────────────────────────────────── _bt_cache: dict = {} BT_TTL = 6 * 3600 # 6 hours # ─── Parameters ────────────────────────────────────────────────────────────── # Using 15m candles: 6 months = ~26,280 bars (15m bars per 6 months) # We cap at 17,280 bars = ~6 months of 15m data (17280 = 6*30*24*4) LOOKBACK_BARS = 17_280 # ~6 months of 15m candles WARMUP_BARS = 100 # need at least this many bars to classify state MAX_HOLD_BARS = 480 # timeout after 5 days (480 × 15m = 5 days) ATR_PERIOD = 14 EMA_SPAN_20 = 20 EMA_SPAN_50 = 50 # Trade level multipliers — MUST match scorer.py _stop_target logic exactly SL_ATR_MULT = 1.2 TP1_ATR_MULT = 1.5 # primary win target TP2_ATR_MULT = 2.5 # ambitious target # ─── Helpers ───────────────────────────────────────────────────────────────── def _ema(arr: list[float], span: int) -> list[float]: k = 2.0 / (span + 1) result = [arr[0]] for v in arr[1:]: result.append(result[-1] * (1 - k) + v * k) return result def _rma(arr: list[float], period: int) -> list[float]: """Wilder RMA (used for ATR).""" k = 1.0 / period result = [arr[0]] for v in arr[1:]: result.append(result[-1] * (1 - k) + v * k) return result def _calc_atr(highs: list, lows: list, closes: list, period: int = 14) -> list[float]: trs = [] for i in range(1, len(closes)): tr = max( highs[i] - lows[i], abs(highs[i] - closes[i-1]), abs(lows[i] - closes[i-1]), ) trs.append(tr) if not trs: return [0.0] seed = sum(trs[:period]) / min(period, len(trs)) rma_vals = _rma([seed] + trs[period:], period) return [0.0] * period + rma_vals def _calc_rsi(closes: list[float], period: int = 14) -> float: if len(closes) < period + 1: return 50.0 deltas = [closes[i] - closes[i-1] for i in range(1, len(closes))] gains = [max(d, 0) for d in deltas[-period:]] losses = [max(-d, 0) for d in deltas[-period:]] ag = sum(gains) / period al = sum(losses) / period if al == 0: return 100.0 if ag == 0: return 0.0 return 100.0 - 100.0 / (1.0 + ag / al) def _signal_score_at(closes: list[float], volumes: list[float], highs: list[float], lows: list[float], atrs: list[float], i: int) -> tuple[int, str]: """5-signal detector, threshold ≥ 4. All signals correlated with trend. S1 EMA spread — trend has clear separation (> 0.5%) S2 RSI confirms — RSI > 50 for long, < 50 for short (moves WITH EMA) S3 Price breakout — close near recent 10-bar high (long) or low (short) S4 Volume spike — current bar > 1.5× 20-bar average S5 ATR expansion — momentum building, ATR ≥ 1.2× avg Direction derived from EMA20 vs EMA50. """ if i < 20: return 0, "long" window = closes[max(0, i - EMA_SPAN_50):i + 1] e20 = _ema(window, EMA_SPAN_20)[-1] e50 = _ema(window, EMA_SPAN_50)[-1] if len(window) >= EMA_SPAN_50 else e20 direction = "long" if e20 >= e50 else "short" score = 0 # S1: EMA spread spread = abs(e20 - e50) / e50 if e50 > 0 else 0 if spread > 0.005: score += 1 # S2: RSI trend confirmation rsi = _calc_rsi(closes[max(0, i - 14):i + 1]) if direction == "long" and rsi > 50: score += 1 if direction == "short" and rsi < 50: score += 1 # S3: Price near recent 10-bar extreme if i >= 10: recent_high = max(highs[i - 10:i]) recent_low = min(lows[i - 10:i]) if direction == "long" and closes[i] >= recent_high * 0.998: score += 1 if direction == "short" and closes[i] <= recent_low * 1.002: score += 1 # S4: Volume spike vol_avg = statistics.mean(volumes[i - 20:i]) if i >= 20 else volumes[i] if vol_avg > 0 and volumes[i] / vol_avg >= 1.5: score += 1 # S5: ATR expansion atr_now = atrs[i] atr_list = [a for a in atrs[i - 20:i] if a > 0] atr_avg = statistics.mean(atr_list) if atr_list else 0 if atr_avg > 0 and atr_now / atr_avg >= 1.2: score += 1 return score, direction # ─── Core walk-forward loop ─────────────────────────────────────────────────── def run_backtest(df, symbol: str = "") -> dict: """Walk forward through 15m df, simulate trades, return stats dict. df must have columns: open, high, low, close, volume (pandas DataFrame). Uses last LOOKBACK_BARS rows. Tracks TWO win conditions: - TP1: 1.5× ATR target hit before SL (matches live trading style) - TP2: 2.5× ATR target hit before SL (ambitious hold) """ import pandas as pd if len(df) > LOOKBACK_BARS: df = df.iloc[-LOOKBACK_BARS:].reset_index(drop=True) closes = [float(x) for x in df["close"].values] highs = [float(x) for x in df["high"].values] if "high" in df.columns else closes lows = [float(x) for x in df["low"].values] if "low" in df.columns else closes volumes = [float(x) for x in df["volume"].values] if "volume" in df.columns else [1.0]*len(closes) atrs = _calc_atr(highs, lows, closes, ATR_PERIOD) trades = [] in_trade = False trade_entry = trade_sl = trade_tp1 = trade_tp2 = 0.0 trade_dir = "long" trade_state = "RANGING" trade_bar = 0 tp1_hit_bar = None # track if TP1 was hit during this trade class _FakeDF: def __init__(self, c, v): import pandas as pd self._df = pd.DataFrame({"close": c, "volume": v}) self.columns = self._df.columns def __getitem__(self, key): return self._df[key] def __len__(self): return len(self._df) i = WARMUP_BARS while i < len(closes) - 1: if not in_trade: sig_score, sig_dir = _signal_score_at(closes, volumes, highs, lows, atrs, i) if sig_score >= 4: try: fake_df = _FakeDF(closes[:i+1], volumes[:i+1]) state, conf, _ = classify_state(fake_df) except Exception: state, conf = "RANGING", 0.5 # State gate: only block if confidence is HIGH and direction conflicts state_ok = True if conf >= 0.70: if state == "BEAR" and sig_dir == "long": state_ok = False if state == "BULL" and sig_dir == "short": state_ok = False if state_ok: entry = closes[i] atr = atrs[i] if atrs[i] > 0 else entry * 0.005 if sig_dir == "long": sl = entry - SL_ATR_MULT * atr tp1 = entry + TP1_ATR_MULT * atr tp2 = entry + TP2_ATR_MULT * atr else: sl = entry + SL_ATR_MULT * atr tp1 = entry - TP1_ATR_MULT * atr tp2 = entry - TP2_ATR_MULT * atr in_trade = True trade_entry = entry trade_sl = sl trade_tp1 = tp1 trade_tp2 = tp2 trade_dir = sig_dir trade_state = state trade_bar = i tp1_hit_bar = None else: hi = highs[i] lo = lows[i] if trade_dir == "long": hit_tp1 = hi >= trade_tp1 hit_tp2 = hi >= trade_tp2 hit_sl = lo <= trade_sl else: hit_tp1 = lo <= trade_tp1 hit_tp2 = lo <= trade_tp2 hit_sl = hi >= trade_sl # Track first TP1 touch (even if we continue holding for TP2) if hit_tp1 and tp1_hit_bar is None: tp1_hit_bar = i bars_held = i - trade_bar timeout = bars_held >= MAX_HOLD_BARS if hit_tp2 or hit_sl or timeout: risk_r = abs(trade_entry - trade_sl) # ── TP1 outcome ─────────────────────────────────────────── if tp1_hit_bar is not None: # TP1 was touched at some point before SL/timeout # Check: was SL hit BEFORE TP1? # We check the bar at tp1_hit_bar for SL as well outcome_tp1 = "win" # But if SL was hit on the same bar as TP1 first touch, check direction # (conservative: if both same bar, credit TP1 win for longs if close > entry) else: # TP1 never reached if timeout: # Timeout — price never hit TP1, grade as loss outcome_tp1 = "loss" else: # SL hit before TP1 ever touched outcome_tp1 = "loss" # ── TP2 outcome ─────────────────────────────────────────── if hit_tp2 and not hit_sl: outcome_tp2 = "win" elif hit_sl and not hit_tp2: outcome_tp2 = "loss" elif hit_tp2 and hit_sl: outcome_tp2 = "win" if (trade_dir == "long" and closes[i] > trade_entry) else "loss" else: # timeout outcome_tp2 = "win" if ( (trade_dir == "long" and closes[i] > trade_tp1) or (trade_dir == "short" and closes[i] < trade_tp1) ) else "loss" # ── R-multiples ─────────────────────────────────────────── if outcome_tp1 == "win": r_mult_tp1 = round(abs(trade_tp1 - trade_entry) / risk_r, 2) if risk_r > 0 else 0 else: r_mult_tp1 = -1.0 if outcome_tp2 == "win": r_mult_tp2 = round(abs(trade_tp2 - trade_entry) / risk_r, 2) if risk_r > 0 else 0 else: r_mult_tp2 = -1.0 trades.append({ "bar": trade_bar, "state": trade_state, "direction": trade_dir, "outcome_tp1": outcome_tp1, "outcome_tp2": outcome_tp2, "r_mult_tp1": r_mult_tp1, "r_mult_tp2": r_mult_tp2, "bars_held": bars_held, "entry": round(trade_entry, 6), "sl": round(trade_sl, 6), "tp1": round(trade_tp1, 6), "tp2": round(trade_tp2, 6), }) in_trade = False tp1_hit_bar = None i += 1 # ── Aggregate ───────────────────────────────────────────────────────── if not trades: return { "symbol": symbol, "total_trades": 0, "win_rate": None, # TP1 win rate (primary) "win_rate_tp1": None, "win_rate_tp2": None, "avg_r": None, "avg_r_tp1": None, "avg_r_tp2": None, "max_drawdown_r": None, "by_state": {}, "by_direction": {}, "equity_curve": [], "recent_trades": [], "lookback_bars": len(closes), "lookback_months": round(len(closes) / (24 * 4 * 30), 1), # 15m bars per month = 24*4*30 "candle_interval": "15m", "win_condition": "TP1 hit before SL (primary) / TP2 hit before SL (secondary)", "ts": time.time(), } wins_tp1 = [t for t in trades if t["outcome_tp1"] == "win"] wins_tp2 = [t for t in trades if t["outcome_tp2"] == "win"] win_rate_tp1 = round(len(wins_tp1) / len(trades), 3) win_rate_tp2 = round(len(wins_tp2) / len(trades), 3) avg_r_tp1 = round(statistics.mean([t["r_mult_tp1"] for t in trades]), 3) avg_r_tp2 = round(statistics.mean([t["r_mult_tp2"] for t in trades]), 3) # Primary win_rate = TP1 (matches user's actual trading style) win_rate = win_rate_tp1 avg_r = avg_r_tp1 # By state (TP1 primary) by_state = {} for state in ALL_STATES: st = [t for t in trades if t["state"] == state] if st: sw1 = [t for t in st if t["outcome_tp1"] == "win"] sw2 = [t for t in st if t["outcome_tp2"] == "win"] by_state[state] = { "trades": len(st), "wins_tp1": len(sw1), "wins_tp2": len(sw2), "win_rate": round(len(sw1) / len(st), 3), # TP1 "win_rate_tp1": round(len(sw1) / len(st), 3), "win_rate_tp2": round(len(sw2) / len(st), 3), "avg_r": round(statistics.mean([t["r_mult_tp1"] for t in st]), 3), } # By direction (TP1 primary) by_dir = {} for d in ("long", "short"): dt = [t for t in trades if t["direction"] == d] if dt: dw1 = [t for t in dt if t["outcome_tp1"] == "win"] dw2 = [t for t in dt if t["outcome_tp2"] == "win"] by_dir[d] = { "trades": len(dt), "wins_tp1": len(dw1), "wins_tp2": len(dw2), "win_rate": round(len(dw1) / len(dt), 3), "win_rate_tp1": round(len(dw1) / len(dt), 3), "win_rate_tp2": round(len(dw2) / len(dt), 3), "avg_r": round(statistics.mean([t["r_mult_tp1"] for t in dt]), 3), } # Equity curve based on TP1 (how most users actually trade) equity = [] cum_r = 0.0 for t in trades: cum_r += t["r_mult_tp1"] equity.append(round(cum_r, 3)) # Max drawdown on TP1 equity curve peak = 0.0 max_dd = 0.0 for e in equity: if e > peak: peak = e dd = peak - e if dd > max_dd: max_dd = dd return { "symbol": symbol, "total_trades": len(trades), "wins_tp1": len(wins_tp1), "wins_tp2": len(wins_tp2), "losses": len(trades) - len(wins_tp1), "win_rate": win_rate, # TP1 (primary — matches user trading style) "win_rate_tp1": win_rate_tp1, "win_rate_tp2": win_rate_tp2, "avg_r": avg_r, "avg_r_tp1": avg_r_tp1, "avg_r_tp2": avg_r_tp2, "max_drawdown_r": round(max_dd, 3), "by_state": by_state, "by_direction": by_dir, "equity_curve": equity[-200:], "recent_trades": trades[-50:], "lookback_bars": len(closes), "lookback_months": round(len(closes) / (24 * 4 * 30), 1), "candle_interval": "15m", "win_condition": "TP1 hit before SL (primary) / TP2 hit before SL (secondary)", "ts": time.time(), } # ─── Paginated 15m kline fetch ──────────────────────────────────────────────── def _fetch_klines_paginated(src, symbol: str, target_bars: int = LOOKBACK_BARS) -> "pd.DataFrame": """Fetch up to `target_bars` of 15m klines by walking backwards in time. 15m interval: each bar = 15 minutes = 900,000 ms 6 months of 15m bars = ~17,280 bars. Each API call returns max 500 bars → need up to 35 calls. Works for BingX, Binance, Bybit. """ import pandas as pd import requests CHUNK = 500 INTERVAL_MS = 900_000 # 15m in milliseconds INTERVAL_STR = "15m" # First call (always works — uses src.klines wrapper) df_base = src.klines(symbol, INTERVAL_STR) frames = [df_base] collected = len(df_base) if collected >= target_bars: return df_base earliest_ms = int(df_base["open_time"].iloc[0]) src_name = type(src).__name__.lower() for _ in range(40): # up to 40 extra pages = 20,000 extra bars if collected >= target_bars: break end_ms = earliest_ms - 1 try: if "bingx" in src_name: r = requests.get( "https://open-api.bingx.com/openApi/swap/v3/quote/klines", params={ "symbol": symbol, "interval": INTERVAL_STR, "limit": str(CHUNK), "endTime": str(end_ms), }, timeout=10, ).json() rows = (r.get("data") or []) if not rows: break chunk = pd.DataFrame(rows) chunk = chunk.rename(columns={"time": "open_time"}) chunk["open_time"] = chunk["open_time"].astype("int64") for c in ("open", "high", "low", "close", "volume"): chunk[c] = chunk[c].astype(float) chunk = chunk[["open_time", "open", "high", "low", "close", "volume"]] elif "binance" in src_name: raw = requests.get( "https://fapi.binance.com/fapi/v1/klines", params={ "symbol": symbol, "interval": INTERVAL_STR, "limit": CHUNK, "endTime": end_ms, }, timeout=10, ).json() if not raw or isinstance(raw, dict): break chunk = pd.DataFrame(raw, columns=[ "open_time","open","high","low","close", "volume","close_time","qv","trades","tb","tq","ig" ]) chunk["open_time"] = chunk["open_time"].astype("int64") for c in ("open","high","low","close","volume"): chunk[c] = chunk[c].astype(float) chunk = chunk[["open_time","open","high","low","close","volume"]] elif "bybit" in src_name: start_ms = end_ms - CHUNK * INTERVAL_MS raw = requests.get( "https://api.bybit.com/v5/market/kline", params={ "category": "linear", "symbol": symbol, "interval": "15", "limit": CHUNK, "start": start_ms, "end": end_ms, }, timeout=10, ).json() rows = (raw.get("result") or {}).get("list") or [] if not rows: break chunk = pd.DataFrame(list(reversed(rows)), columns=["open_time","open","high","low","close","volume","turnover"]) chunk["open_time"] = chunk["open_time"].astype("int64") for c in ("open","high","low","close","volume"): chunk[c] = chunk[c].astype(float) chunk = chunk[["open_time","open","high","low","close","volume"]] else: break if len(chunk) == 0: break frames.append(chunk) collected += len(chunk) earliest_ms = int(chunk["open_time"].iloc[0]) except Exception: break if not frames: return df_base combined = pd.concat(frames, ignore_index=True) combined = combined.drop_duplicates("open_time").sort_values("open_time").reset_index(drop=True) # Drop still-open (live) candle import time as _time now_ms = int(_time.time() * 1000) combined = combined[combined["open_time"] + INTERVAL_MS <= now_ms].reset_index(drop=True) return combined.tail(target_bars).reset_index(drop=True) def backtest_symbol(src, symbol: str) -> dict: """Fetch 6 months of 15m klines (paginated) and run walk-forward backtest. Cached for BT_TTL seconds. """ cached = _bt_cache.get(symbol) if cached and time.time() - cached["ts"] < BT_TTL: return cached["result"] try: df = _fetch_klines_paginated(src, symbol, target_bars=LOOKBACK_BARS) result = run_backtest(df, symbol=symbol) _bt_cache[symbol] = {"result": result, "ts": time.time()} return result except Exception as e: err = { "symbol": symbol, "total_trades": 0, "win_rate": None, "win_rate_tp1": None, "win_rate_tp2": None, "avg_r": None, "max_drawdown_r": None, "lookback_months": None, "candle_interval": "15m", "error": str(e)[:120], "ts": time.time(), } _bt_cache[symbol] = {"result": err, "ts": time.time()} return err