""" services/backtest_engine.py — Walk-forward backtesting engine. Uses the same strategy logic as the live scanner with no lookahead. bt_* functions are self-contained (no dependency on data/indicators.py) to ensure exact parity with what the backtester computes vs live. """ import numpy as np import pandas as pd import streamlit as st from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta, timezone from oandapyV20 import API import oandapyV20.endpoints.instruments as instruments_ep from config import ( EMA_FAST, EMA_SLOW, SMA_TREND, RSI_PERIOD, RSI_MA_PERIOD, MACD_FAST, MACD_SLOW, MACD_SIGNAL, ADX_PERIOD, VOL_MA, RSI_OB, RSI_OS, ADX_MIN, SR_LOOKBACK, SR_ZONE_PIPS, STRUCT_LOOKBACK, MTF_MIN_SCORE, MIN_RR, SL_BUFFER_PIPS, MIN_SL_PIPS, FALLBACK_SL_PIPS, FALLBACK_TP_PIPS, ) # ── Instrument helpers ──────────────────────────────── def bt_pip(inst: str) -> float: return 0.01 if ("_JPY" in inst or inst == "XAU_USD") else 0.0001 def bt_digs(inst: str) -> int: if "_JPY" in inst: return 3 if inst == "XAU_USD": return 2 return 5 # ── Indicators (self-contained for backtest) ────────── def bt_add_all(df: pd.DataFrame, include_trend: bool = False) -> pd.DataFrame: """Apply all indicators — mirrors data/indicators.py add_all exactly.""" df = df.copy() df["ma_fast"] = df["close"].ewm(span=EMA_FAST, adjust=False).mean() df["ma_slow"] = df["close"].ewm(span=EMA_SLOW, adjust=False).mean() if include_trend: df["ma_trend"] = df["close"].rolling(SMA_TREND).mean() df["ema_stack_bull"] = (df["ma_fast"] > df["ma_slow"]) & (df["ma_slow"] > df["ma_trend"]) df["ema_stack_bear"] = (df["ma_fast"] < df["ma_slow"]) & (df["ma_slow"] < df["ma_trend"]) df["dist_slow"] = df["close"] - df["ma_slow"] conds = [ (df["close"] > df["ma_fast"]) & (df["ma_fast"] > df["ma_slow"]), (df["close"] < df["ma_fast"]) & (df["ma_fast"] < df["ma_slow"]), (df["close"] > df["ma_slow"]) & (df["close"] < df["ma_fast"]), (df["close"] < df["ma_slow"]) & (df["close"] > df["ma_fast"]), ] df["price_zone"] = np.select(conds, ["above_both","below_both","between_pb_up","between_pb_dn"], default="mixed") df["cross"] = 0 df.loc[(df["ma_fast"] > df["ma_slow"]) & (df["ma_fast"].shift(1) <= df["ma_slow"].shift(1)), "cross"] = 1 df.loc[(df["ma_fast"] < df["ma_slow"]) & (df["ma_fast"].shift(1) >= df["ma_slow"].shift(1)), "cross"] = -1 delta = df["close"].diff() ag = delta.clip(lower=0).ewm(alpha=1/RSI_PERIOD, min_periods=RSI_PERIOD, adjust=False).mean() al = (-delta.clip(upper=0)).ewm(alpha=1/RSI_PERIOD, min_periods=RSI_PERIOD, adjust=False).mean() df["rsi"] = 100 - (100 / (1 + ag / al.replace(0, np.nan))) df["rsi_ma"] = df["rsi"].rolling(RSI_MA_PERIOD).mean() df["rsi_above_ma"] = df["rsi"] > df["rsi_ma"] ef = df["close"].ewm(span=MACD_FAST, adjust=False).mean() es = df["close"].ewm(span=MACD_SLOW, adjust=False).mean() df["macd"] = ef - es df["macd_sig"] = df["macd"].ewm(span=MACD_SIGNAL, adjust=False).mean() df["macd_hist"] = df["macd"] - df["macd_sig"] hi, lo, cl = df["high"], df["low"], df["close"] pdm = hi.diff().clip(lower=0); ndm = (-lo.diff()).clip(lower=0) pdm[pdm < ndm] = 0; ndm[ndm < pdm] = 0 tr = pd.concat([hi-lo, (hi-cl.shift()).abs(), (lo-cl.shift()).abs()], axis=1).max(axis=1) a = 1 / ADX_PERIOD atr = tr.ewm(alpha=a, min_periods=ADX_PERIOD, adjust=False).mean() pdi = 100 * pdm.ewm(alpha=a, min_periods=ADX_PERIOD, adjust=False).mean() / atr ndi = 100 * ndm.ewm(alpha=a, min_periods=ADX_PERIOD, adjust=False).mean() / atr dx = 100 * (pdi - ndi).abs() / (pdi + ndi).replace(0, np.nan) df["adx"] = dx.ewm(alpha=a, min_periods=ADX_PERIOD, adjust=False).mean() df["plus_di"] = pdi; df["minus_di"] = ndi df["vol_ma"] = df["tick_volume"].rolling(VOL_MA).mean() df["vol_ratio"] = df["tick_volume"] / df["vol_ma"] return df # ── Swing / S&R helpers ─────────────────────────────── def bt_detect_swings(df: pd.DataFrame, n: int = 3) -> pd.DataFrame: df = df.copy() df["swing_high"] = df["high"][(df["high"] == df["high"].rolling(2*n+1, center=True).max())] df["swing_low"] = df["low"] [(df["low"] == df["low"].rolling(2*n+1, center=True).min())] return df def bt_find_sr(df: pd.DataFrame, instrument: str, n: int = 8) -> list: df = bt_detect_swings(df) raw = pd.concat([df.iloc[-SR_LOOKBACK:]["swing_high"].dropna(), df.iloc[-SR_LOOKBACK:]["swing_low"].dropna()]).sort_values().values if not len(raw): return [] zone = SR_ZONE_PIPS * bt_pip(instrument) clusters: list = []; cur = [raw[0]] for p in raw[1:]: if p - cur[-1] <= zone: cur.append(p) else: clusters.append(cur); cur = [p] clusters.append(cur); clusters.sort(key=len, reverse=True) return [np.mean(c) for c in clusters[:n]] # ── Analysis functions ──────────────────────────────── def bt_analyse_htf(df: pd.DataFrame) -> int: row = df.iloc[-1] ema_bull = row["ma_fast"] > row["ma_slow"] pr_bull = row["close"] > row["ma_fast"] has_trend = "ma_trend" in df.columns and not pd.isna(row.get("ma_trend", float("nan"))) stack_bull = bool(row.get("ema_stack_bull", False)) if has_trend else ema_bull stack_bear = bool(row.get("ema_stack_bear", False)) if has_trend else not ema_bull bull = sum([ema_bull, stack_bull, pr_bull]) bear = sum([not ema_bull, stack_bear, not pr_bull]) if bull >= 2 and bull > bear: return 1 if bear >= 2 and bear > bull: return -1 return 0 def bt_mtf_score(df: pd.DataFrame, bias: int, instrument: str) -> tuple[int, int]: if bias == 0: return 0, 7 row = df.iloc[-1]; score = 0 score += int(row["ma_fast"] > row["ma_slow"] if bias == 1 else row["ma_fast"] < row["ma_slow"]) has_stack = "ema_stack_bull" in df.columns if has_stack: score += int(bool(row.get("ema_stack_bull", False)) if bias == 1 else bool(row.get("ema_stack_bear", False))) zone_ok = row["price_zone"] in (("above_both","between_pb_up") if bias==1 else ("below_both","between_pb_dn")) score += int(zone_ok) score += int(row["rsi"] < RSI_OB if bias == 1 else row["rsi"] > RSI_OS) df_sw = bt_detect_swings(df); r = df_sw.iloc[-STRUCT_LOOKBACK:] sh = r["swing_high"].dropna(); sl = r["swing_low"].dropna() if len(sh) >= 2 and len(sl) >= 2: lh = sh.iloc[-2:].values; ll = sl.iloc[-2:].values if bias == 1: score += int(lh[-1] > lh[-2] and ll[-1] > ll[-2]) else: score += int(lh[-1] < lh[-2] and ll[-1] < ll[-2]) r2 = df.iloc[-15:]; row2 = df.iloc[-1] imp = r2[r2["close"] > r2["open"]] if bias == 1 else r2[r2["close"] < r2["open"]] pb = r2[r2["close"] < r2["open"]] if bias == 1 else r2[r2["close"] > r2["open"]] if not imp.empty and not pb.empty: score += int(imp["tick_volume"].mean() > pb["tick_volume"].mean() and row2["vol_ratio"] >= 0.8) prev = df.iloc[-2]; adx_ok = row["adx"] > ADX_MIN if bias == 1: score += int(row["macd"] > row["macd_sig"] and row["macd_hist"] > prev["macd_hist"] and adx_ok and row["plus_di"] > row["minus_di"]) else: score += int(row["macd"] < row["macd_sig"] and row["macd_hist"] < prev["macd_hist"] and adx_ok and row["minus_di"] > row["plus_di"]) max_score = 8 if has_stack else 7 return score, max_score def bt_ltf_trigger(df: pd.DataFrame, bias: int) -> tuple[bool, bool]: if bias == 0 or len(df) < 3: return False, False row = df.iloc[-1]; prev = df.iloc[-2]; recent = df.iloc[-10:] if bias == 1: cross_recent = any(df["cross"].iloc[-5:] == 1) cross_aligned = row["ma_fast"] > row["ma_slow"] cross_extended = any(df["cross"].iloc[-10:] == 1) cross_ok = cross_recent or (cross_aligned and cross_extended) adx_strong = row["adx"] > 30 rsi_pullback = (recent["rsi"] < RSI_OS+10).any() and row["rsi"] > prev["rsi"] and row["rsi"] < RSI_OB rsi_trend = adx_strong and row["rsi"] < 55 and row["rsi"] > prev["rsi"] rsi_momentum = row["rsi"] > 55 and row["rsi"] > prev["rsi"] and row["rsi"] < RSI_OB rsi_ok = rsi_pullback or rsi_trend or rsi_momentum else: cross_recent = any(df["cross"].iloc[-5:] == -1) cross_aligned = row["ma_fast"] < row["ma_slow"] cross_extended = any(df["cross"].iloc[-10:] == -1) cross_ok = cross_recent or (cross_aligned and cross_extended) adx_strong = row["adx"] > 30 rsi_pullback = (recent["rsi"] > RSI_OB-10).any() and row["rsi"] < prev["rsi"] and row["rsi"] > RSI_OS rsi_trend = adx_strong and row["rsi"] > 45 and row["rsi"] < prev["rsi"] rsi_oversold = adx_strong and row["rsi"] < RSI_OS + 5 rsi_momentum = row["rsi"] < 45 and row["rsi"] < prev["rsi"] and row["rsi"] > RSI_OS rsi_ok = rsi_pullback or rsi_trend or rsi_oversold or rsi_momentum return cross_ok, rsi_ok def bt_count_patterns(df_mtf: pd.DataFrame, df_ltf: pd.DataFrame, bias: int) -> tuple[int, float | None]: score = 0; neckline = None try: df_sw = bt_detect_swings(df_mtf); rec = df_sw.iloc[-60:] if bias == -1: peaks = rec["swing_high"].dropna() if len(peaks) >= 3: l, h, r = peaks.values[-3], peaks.values[-2], peaks.values[-1] if h > l and h > r and abs(l-r)/h < 0.30: troughs = rec["swing_low"].dropna() if len(troughs) >= 2: nl = np.mean(troughs.values[-2:]) if df_mtf.iloc[-1]["close"] < nl: score += 1; neckline = nl else: troughs = rec["swing_low"].dropna() if len(troughs) >= 3: l, h, r = troughs.values[-3], troughs.values[-2], troughs.values[-1] if h < l and h < r and abs(l-r)/abs(h) < 0.30: peaks = rec["swing_high"].dropna() if len(peaks) >= 2: nl = np.mean(peaks.values[-2:]) if df_mtf.iloc[-1]["close"] > nl: score += 1; neckline = nl except Exception: pass try: if len(df_ltf) >= 20: pole = df_ltf.iloc[-20:-8]; flag = df_ltf.iloc[-8:] if not pole.empty and not flag.empty: pm = pole["close"].iloc[-1] - pole["close"].iloc[0] pct = abs(pm) / pole["close"].iloc[0] * 100 fr = (flag["high"].max() - flag["low"].min()) / abs(pm) if abs(pm) > 0 else 1 if pct >= 0.5 and fr <= 0.65: price = df_ltf.iloc[-1]["close"] if bias == 1 and pm > 0 and price > flag["high"].max(): score += 1 elif bias == -1 and pm < 0 and price < flag["low"].min(): score += 1 except Exception: pass return score, neckline def bt_calc_sl_tp(df_ltf, df_mtf, instrument, direction, entry, neckline=None): p = bt_pip(instrument); d = bt_digs(instrument); buf = SL_BUFFER_PIPS * p sw = bt_detect_swings(df_ltf); rec = sw.iloc[-STRUCT_LOOKBACK:] if direction == "buy": lows = rec["swing_low"].dropna(); valid = lows[lows < entry].sort_index() sl = round(valid.iloc[-1] - buf, d) if not valid.empty else round(entry - FALLBACK_SL_PIPS*p, d) # No fixed minimum — structure-based only else: highs = rec["swing_high"].dropna(); valid = highs[highs > entry].sort_index() sl = round(valid.iloc[-1] + buf, d) if not valid.empty else round(entry + FALLBACK_SL_PIPS*p, d) # No fixed minimum — structure-based only sl_pips = abs(entry - sl) / p # TP1 = 1:1 R:R, TP2 = next S&R, TP3 = trail tp1 = round(entry + (entry - sl) if direction == "buy" else entry - (sl - entry), d) if neckline and direction == "buy" and neckline > entry: tp2 = round(neckline, d) elif neckline and direction == "sell" and neckline < entry: tp2 = round(neckline, d) else: levels = bt_find_sr(df_mtf, instrument) if direction == "buy": tgts = sorted([l for l in levels if l > entry + sl_pips*p*2]) tp2 = round(tgts[0], d) if tgts else round(entry + FALLBACK_TP_PIPS*p, d) else: tgts = sorted([l for l in levels if l < entry - sl_pips*p*2], reverse=True) tp2 = round(tgts[0], d) if tgts else round(entry - FALLBACK_TP_PIPS*p, d) tp2_pips = abs(tp2 - entry) / p tp1_pips = abs(tp1 - entry) / p rr = round(tp2_pips / sl_pips, 2) if sl_pips > 0 else 0 return sl, tp1, tp2, round(sl_pips, 1), round(tp1_pips, 1), round(tp2_pips, 1), rr # ── Data fetching ───────────────────────────────────── @st.cache_data(ttl=86400, show_spinner=False) def bt_fetch_candles_cached(token, environment, instrument, granularity, date_from, date_to): """Cached historical candle fetch — 24h TTL (historical data never changes).""" client = API(access_token=token, environment=environment) return _bt_fetch_raw(client, instrument, granularity, date_from, date_to) def _bt_fetch_raw(client, instrument, granularity, date_from, date_to): start = datetime.strptime(date_from, "%Y-%m-%d").replace(tzinfo=timezone.utc) end = datetime.strptime(date_to, "%Y-%m-%d").replace(tzinfo=timezone.utc) chunk_map = {"D": 400, "H4": 500, "H1": 180, "M30": 90, "M15": 40} chunk_days = chunk_map.get(granularity, 180) all_rows: list = [] while start < end: chunk_end = min(end, start + timedelta(days=chunk_days)) params = {"granularity": granularity, "from": start.strftime("%Y-%m-%dT%H:%M:%SZ"), "to": chunk_end.strftime("%Y-%m-%dT%H:%M:%SZ"), "price": "M"} try: r = instruments_ep.InstrumentsCandles(instrument, params=params) client.request(r) candles = r.response.get("candles", []) except Exception: break if not candles: start = chunk_end + timedelta(seconds=1) continue for c in candles: if not c.get("complete", True): continue m = c["mid"] all_rows.append({"time": pd.to_datetime(c["time"]), "open": float(m["o"]), "high": float(m["h"]), "low": float(m["l"]), "close": float(m["c"]), "tick_volume": int(c["volume"])}) last = pd.to_datetime(candles[-1]["time"]).to_pydatetime() if last.tzinfo is None: last = last.replace(tzinfo=timezone.utc) start = last + timedelta(seconds=1) if not all_rows: return pd.DataFrame() return pd.DataFrame(all_rows).drop_duplicates("time").set_index("time").sort_index() def bt_fetch_all_parallel(token, environment, instrument, date_from, date_to, progress_cb=None, include_30m=False): """Download D + H4 + H1 (+ M30) simultaneously using threads — 3× faster.""" timeframes = ["D", "H4", "H1", "M30"] if include_30m else ["D", "H4", "H1"] results: dict = {} def fetch_one(tf): return tf, bt_fetch_candles_cached(token, environment, instrument, tf, date_from, date_to) n_tfs = len(timeframes) tfs_label = "D1 + H4 + H1 + 30M" if include_30m else "D1 + H4 + H1" if progress_cb: progress_cb(0.05, f"Downloading {instrument} ({tfs_label}) · checking cache...") with ThreadPoolExecutor(max_workers=4) as executor: futures = {executor.submit(fetch_one, tf): tf for tf in timeframes} done = 0 for future in as_completed(futures): tf, df = future.result() results[tf] = df; done += 1 cached_str = "⚡ cached" if len(df) > 0 else "downloaded" if progress_cb: progress_cb(0.05 + 0.40*(done/n_tfs), f"{instrument} {tf} — {len(df):,} candles {cached_str} ({done}/{n_tfs})") return (results.get("D", pd.DataFrame()), results.get("H4", pd.DataFrame()), results.get("H1", pd.DataFrame()), results.get("M30", pd.DataFrame())) # ── Main backtest loop ──────────────────────────────── def run_backtest(client, token, environment, instrument, date_from, date_to, progress_cb=None, ltf="H1"): """Walk-forward backtest — same logic as live scanner, zero lookahead.""" use_30m = (ltf == "M30") _start = datetime.strptime(date_from, "%Y-%m-%d") warmup_from = (_start - timedelta(days=300)).strftime("%Y-%m-%d") df_d1_raw, df_h4_raw, df_h1_raw, df_m30_raw = bt_fetch_all_parallel( token, environment, instrument, warmup_from, date_to, progress_cb, include_30m=use_30m) _bt_start = pd.Timestamp(date_from, tz="UTC") df_h4_raw = df_h4_raw[df_h4_raw.index >= _bt_start] df_h1_raw_full = df_h1_raw.copy() df_h1_raw = df_h1_raw[df_h1_raw.index >= _bt_start] if use_30m and not df_m30_raw.empty: df_m30_raw = df_m30_raw[df_m30_raw.index >= _bt_start] if df_d1_raw.empty or df_h4_raw.empty or df_h1_raw.empty: return [], f"No data for {instrument}" if use_30m and df_m30_raw.empty: return [], f"No 30M data for {instrument}" if progress_cb: progress_cb(0.55, "Computing indicators...") df_d1 = bt_add_all(df_d1_raw, include_trend=True) df_h4 = bt_add_all(df_h4_raw, include_trend=True) df_h1 = bt_add_all(df_h1_raw) df_m30 = bt_add_all(df_m30_raw) if use_30m else pd.DataFrame() df_entry = df_m30 if use_30m else df_h1 warmup_date = df_d1.index[SMA_TREND] if len(df_d1) > SMA_TREND else df_d1.index[min(50, len(df_d1)-1)] trades: list = [] open_trade: dict | None = None balance = 10000.0 n_bars = len(df_entry) ltf_label = "30M" if use_30m else "1H" if progress_cb: progress_cb(0.60, f"Running walk-forward simulation on {ltf_label}...") for i in range(50, n_bars): bar_time = df_entry.index[i] bar = df_entry.iloc[i] if bar_time < warmup_date: continue if bar_time.hour not in range(7, 21): if open_trade: open_trade, closed = bt_check_exit(open_trade, bar, bar_time, instrument) if closed: balance += bt_calc_pnl(closed, instrument) closed["balance"] = round(balance, 2) trades.append(closed); open_trade = None continue if open_trade: open_trade, closed = bt_check_exit(open_trade, bar, bar_time, instrument) if closed: balance += bt_calc_pnl(closed, instrument) closed["balance"] = round(balance, 2) trades.append(closed); open_trade = None else: continue d1 = df_d1[df_d1.index <= bar_time] h4 = df_h4[df_h4.index <= bar_time] entry_slice = df_entry.iloc[:i+1] h1 = df_h1[df_h1.index <= bar_time] if use_30m else entry_slice if len(d1) < 10 or len(h4) < 10 or len(entry_slice) < 10: continue bias = bt_analyse_htf(d1) mtf_score, mtf_max = bt_mtf_score(h4, bias, instrument) cross_ok, rsi_ok = bt_ltf_trigger(entry_slice, bias) pat_score, neckline = bt_count_patterns(h4, entry_slice, bias) if bias == 0: continue if mtf_score < MTF_MIN_SCORE: continue if not cross_ok: continue if not rsi_ok: continue direction = "BUY" if bias == 1 else "SELL" entry = bar["close"] sl, tp1, tp2, sl_pips, tp1_pips, tp2_pips, rr = bt_calc_sl_tp( entry_slice, h4, instrument, direction.lower(), entry, neckline) if rr < MIN_RR: continue open_trade = { "instrument": instrument, "direction": direction, "entry": entry, "entry_time": bar_time, "sl": sl, "tp": tp1, "sl_pips": sl_pips, "tp_pips": tp1_pips, "tp1": tp1, "tp2": tp2, "tp1_pips": tp1_pips, "tp2_pips": tp2_pips, "rr": rr, "mtf_score": mtf_score, "mtf_max": mtf_max, "pat_score": pat_score, "cross_ok": cross_ok, "rsi_ok": rsi_ok, "bias": bias, "ltf": ltf_label, } if progress_cb and i % 100 == 0: progress_cb(0.60 + 0.35*(i/n_bars), f"Simulating... bar {i}/{n_bars} | {bar_time.strftime('%Y-%m-%d')} | {len(trades)} trades") if open_trade: last = df_h1.iloc[-1]; last_time = df_h1.index[-1] ep = last["close"] pnl_pips = (ep - open_trade["entry"]) / bt_pip(instrument) * (1 if open_trade["direction"] == "BUY" else -1) balance += bt_calc_pnl({**open_trade, "pnl_pips": pnl_pips}, instrument) trades.append({**open_trade, "exit_price": ep, "exit_time": last_time, "exit_reason": "end", "pnl_pips": round(pnl_pips, 1), "balance": round(balance, 2)}) if progress_cb: msg = f"Done! {len(trades)} trades found." if trades else "Done — 0 trades. Try a longer date range." progress_cb(1.0, msg) return trades, None # ── Exit / P&L helpers ──────────────────────────────── def bt_check_exit(trade: dict, bar, bar_time, instrument) -> tuple: sl = trade["sl"]; tp = trade["tp"]; d = trade["direction"] hit_sl = bar["low"] <= sl if d == "BUY" else bar["high"] >= sl hit_tp = bar["high"] >= tp if d == "BUY" else bar["low"] <= tp if not hit_sl and not hit_tp: return trade, None if hit_sl and hit_tp: hit_tp = False exit_price = sl if hit_sl else tp exit_reason = "SL" if hit_sl else "TP" pnl_pips = ((exit_price - trade["entry"]) / bt_pip(instrument) if d == "BUY" else (trade["entry"] - exit_price) / bt_pip(instrument)) return None, {**trade, "exit_price": exit_price, "exit_time": bar_time, "exit_reason": exit_reason, "pnl_pips": round(pnl_pips, 1)} def bt_calc_pnl(trade: dict, instrument: str) -> float: pips = trade["pnl_pips"]; lots = 0.01 if "_JPY" in instrument: return pips * lots * 100 if instrument == "XAU_USD": return pips * lots * 1.0 return pips * lots * 10 # ── Analytics ───────────────────────────────────────── def bt_calc_stats(trades: list) -> dict: if not trades: return {} df = pd.DataFrame(trades) closed = df[df["exit_reason"].notna() & (df["exit_reason"] != "end")] if closed.empty: return {} wins = closed[closed["pnl_pips"] > 0] losses = closed[closed["pnl_pips"] <= 0] gp = wins["pnl_pips"].sum() if not wins.empty else 0 gl = abs(losses["pnl_pips"].sum()) if not losses.empty else 1 balances = [10000.0] + [t.get("balance", 10000) for t in trades if "balance" in t] peak = balances[0]; max_dd = 0 for b in balances: peak = max(peak, b) max_dd = max(max_dd, (peak - b) / peak * 100) return { "total": len(closed), "wins": len(wins), "losses": len(losses), "win_rate": round(len(wins) / len(closed) * 100, 1) if len(closed) else 0, "avg_win": round(wins["pnl_pips"].mean(), 1) if not wins.empty else 0, "avg_loss": round(abs(losses["pnl_pips"].mean()), 1) if not losses.empty else 0, "rr_actual": round(wins["pnl_pips"].mean() / abs(losses["pnl_pips"].mean()), 2) if not wins.empty and not losses.empty else 0, "pf": round(gp / gl, 2) if gl > 0 else 0, "max_dd": round(max_dd, 1), "tp_hits": len(closed[closed["exit_reason"] == "TP"]), "sl_hits": len(closed[closed["exit_reason"] == "SL"]), "final_bal": round(balances[-1], 2), "return_pct": round((balances[-1] - 10000) / 10000 * 100, 1), } def bt_setup_probability(trades: list) -> pd.DataFrame: if not trades: return pd.DataFrame() df = pd.DataFrame(trades) closed = df[df["exit_reason"].notna() & (df["exit_reason"] != "end")].copy() if closed.empty: return pd.DataFrame() closed["win"] = closed["pnl_pips"] > 0 rows: list = [] for score in sorted(closed["mtf_score"].unique()): grp = closed[closed["mtf_score"] == score] rows.append({"Condition": f"MTF Score = {score}/{grp['mtf_max'].iloc[0] if 'mtf_max' in grp else 8}", "Trades": len(grp), "Wins": grp["win"].sum(), "Win %": round(grp["win"].mean()*100, 1), "Avg PnL (pips)": round(grp["pnl_pips"].mean(), 1)}) for pc in sorted(closed["pat_score"].unique()): grp = closed[closed["pat_score"] == pc] rows.append({"Condition": f"Patterns confirmed = {pc}", "Trades": len(grp), "Wins": grp["win"].sum(), "Win %": round(grp["win"].mean()*100, 1), "Avg PnL (pips)": round(grp["pnl_pips"].mean(), 1)}) for d in closed["direction"].unique(): grp = closed[closed["direction"] == d] rows.append({"Condition": f"Direction = {d}", "Trades": len(grp), "Wins": grp["win"].sum(), "Win %": round(grp["win"].mean()*100, 1), "Avg PnL (pips)": round(grp["pnl_pips"].mean(), 1)}) return pd.DataFrame(rows) def bt_monthly_breakdown(trades: list) -> pd.DataFrame: if not trades: return pd.DataFrame() df = pd.DataFrame(trades) closed = df[df["exit_reason"].notna() & (df["exit_reason"] != "end")].copy() if closed.empty: return pd.DataFrame() closed["month"] = pd.to_datetime(closed["entry_time"]).dt.to_period("M").astype(str) monthly = closed.groupby("month").agg( Trades=("pnl_pips","count"), Wins=("pnl_pips", lambda x: (x>0).sum()), PnL_pips=("pnl_pips","sum")).reset_index() monthly["Win %"] = (monthly["Wins"] / monthly["Trades"] * 100).round(1) monthly["PnL_pips"] = monthly["PnL_pips"].round(1) monthly.columns = ["Month","Trades","Wins","PnL (pips)","Win %"] return monthly def bt_simulate_from_journal(token: str, environment: str, date_from: str, date_to: str, starting_balance: float = 10000.0, risk_pct: float = 1.0) -> pd.DataFrame: """ Simulate equity curve from journal signals using backtesting logic. Uses journal entries (entry/sl/tp1/tp2) to compute outcome and PnL. Returns DataFrame with columns: date, pnl_pips, pnl_pct, balance, outcome, pair, grade. """ from journal.trade_journal import get_conn, init_db init_db() conn = get_conn() try: df = pd.read_sql( """SELECT signal_time_utc, pair, direction, entry, sl, tp1, tp2, grade, outcome, pnl, session_note, setup_name FROM trades WHERE entry IS NOT NULL AND sl IS NOT NULL AND signal_time_utc >= ? AND signal_time_utc <= ? ORDER BY signal_time_utc ASC""", conn, params=(date_from, date_to), ) except Exception: df = pd.DataFrame() finally: conn.close() if df.empty: return pd.DataFrame() balance = starting_balance rows = [] for _, trade in df.iterrows(): entry = float(trade["entry"] or 0) sl = float(trade["sl"] or 0) tp1 = float(trade["tp1"] or 0) pair = str(trade["pair"] or "") direction = str(trade["direction"] or "").upper() outcome = str(trade["outcome"] or "") manual_pnl = trade.get("pnl") if not entry or not sl or entry == sl: continue pip = bt_pip(pair) sl_pips = abs(entry - sl) / pip if pip else 0 if sl_pips == 0: continue risk_amount = balance * (risk_pct / 100) # Use manual outcome if recorded, else use tp1/sl levels if outcome == "TP": tp_use = tp1 if tp1 else entry pnl_pips = abs(tp_use - entry) / pip if direction == "BUY" else abs(entry - tp_use) / pip pnl_pct = round((pnl_pips / sl_pips) * risk_pct, 3) elif outcome == "SL": pnl_pips = -sl_pips pnl_pct = -risk_pct elif outcome == "BE": pnl_pips = 0 pnl_pct = 0 elif manual_pnl is not None: pnl_pips = float(manual_pnl) pnl_pct = round((pnl_pips / sl_pips) * risk_pct, 3) if sl_pips else 0 else: continue # open trade — skip balance += balance * (pnl_pct / 100) rows.append({ "date": trade["signal_time_utc"], "pair": pair, "grade": trade.get("grade", ""), "setup": trade.get("setup_name", ""), "session": trade.get("session_note", ""), "direction": direction, "outcome": outcome, "pnl_pips": round(pnl_pips, 1), "pnl_pct": round(pnl_pct, 3), "balance": round(balance, 2), }) return pd.DataFrame(rows)