Spaces:
Running
Running
| """ | |
| 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 ───────────────────────────────────── | |
| 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) | |