backtest.py

"""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