trumpsignal-backend/scripts/backtest_signals.py

"""
Backtest: Reversal + Breakout signals on Binance Futures 5m klines
Covers 2023-01-01 to present across SOL, ETH, AVAX, LINK, DOGE
"""

import io
import time
import zipfile
import subprocess
import warnings
import pandas as pd
import numpy as np
from datetime import datetime, timezone
from dateutil.relativedelta import relativedelta
from tabulate import tabulate

warnings.filterwarnings("ignore")

# ── Config ────────────────────────────────────────────────────────────────────

SYMBOLS = ["ETHUSDT", "LINKUSDT", "AVAXUSDT", "SOLUSDT", "DOGEUSDT"]
BTC_SYMBOL = "BTCUSDT"
START_TS = int(datetime(2023, 1, 1, tzinfo=timezone.utc).timestamp() * 1000)
END_TS   = int(datetime(2025, 12, 31, tzinfo=timezone.utc).timestamp() * 1000)

STOP_LOSS   = -0.03   # -3%
TAKE_PROFIT =  0.035  # +3.5%
MAX_HOLD_CANDLES = 288  # 24h at 5m

TREND_MA_PERIOD = 48    # 4h trend filter (48 x 5m = 4h)
BTC_TREND_MA    = 288   # BTC 24h trend filter (288 x 5m = 24h)

# Signal params
REVERSAL_TAKER_BUY_THRESH  = 0.65
REVERSAL_PREV_TAKER_MAX    = 0.45
REVERSAL_MA_PERIOD         = 20
REVERSAL_4H_DECLINE        = -0.05

BREAKOUT_BB_PERIOD         = 20
BREAKOUT_BB_SQUEEZE_PCT    = 20   # bottom 20% of BB width history (60 candles)
BREAKOUT_VOLUME_MULT       = 2.5
BREAKOUT_TAKER_BUY_THRESH  = 0.60

DATA_BASE  = "https://data.binance.vision/data/futures/um/monthly/klines"
CACHE_DIR  = "/tmp/binance_klines_cache"
KLINE_COLS = [
    "open_time", "open", "high", "low", "close", "volume",
    "close_time", "quote_volume", "trades",
    "taker_buy_base", "taker_buy_quote", "ignore"
]

# ── Data fetch ────────────────────────────────────────────────────────────────

def _fetch_month(symbol: str, year: int, month: int):
    """Download one monthly zip from data.binance.vision and return DataFrame."""
    url = f"{DATA_BASE}/{symbol}/5m/{symbol}-5m-{year}-{month:02d}.zip"
    for attempt in range(3):
        result = subprocess.run(
            ["curl", "-s", "-x", "http://127.0.0.1:7890",
             "--max-time", "120", "--output", "-", url],
            capture_output=True,
        )
        if result.returncode == 0 and result.stdout:
            try:
                with zipfile.ZipFile(io.BytesIO(result.stdout)) as zf:
                    csv_name = zf.namelist()[0]
                    with zf.open(csv_name) as f:
                        df = pd.read_csv(f, names=KLINE_COLS, skiprows=1)
                return df
            except Exception:
                pass
        time.sleep(2 ** attempt)
    return None


def fetch_klines(symbol: str, start_ms: int, end_ms: int) -> pd.DataFrame:
    import os
    os.makedirs(CACHE_DIR, exist_ok=True)
    cache_file = f"{CACHE_DIR}/{symbol}.pkl"

    start_dt = datetime.fromtimestamp(start_ms / 1000, tz=timezone.utc)
    end_dt   = datetime.fromtimestamp(end_ms   / 1000, tz=timezone.utc)

    # Load from cache if available
    if os.path.exists(cache_file):
        print(f"  Loading {symbol} from cache...", end="", flush=True)
        df = pd.read_pickle(cache_file)
        df = df[(df.index >= pd.Timestamp(start_dt)) & (df.index <= pd.Timestamp(end_dt))]
        print(f" {len(df)} candles (cached)")
        return df

    frames = []
    cur = start_dt.replace(day=1)
    print(f"  Fetching {symbol}", end="", flush=True)

    while cur <= end_dt:
        df = _fetch_month(symbol, cur.year, cur.month)
        if df is not None:
            frames.append(df)
            print(".", end="", flush=True)
        else:
            print("x", end="", flush=True)
        cur += relativedelta(months=1)

    if not frames:
        print(f" FAILED")
        return pd.DataFrame()

    df = pd.concat(frames, ignore_index=True)
    for col in ["open", "high", "low", "close", "volume", "taker_buy_base"]:
        df[col] = pd.to_numeric(df[col], errors="coerce")
    df["open_time"] = pd.to_datetime(df["open_time"], unit="ms", utc=True)
    df = df.set_index("open_time").sort_index()

    # Save to cache before filtering
    df.to_pickle(cache_file)

    df = df[(df.index >= pd.Timestamp(start_dt)) & (df.index <= pd.Timestamp(end_dt))]
    print(f" {len(df)} candles")
    return df


# ── Indicators ────────────────────────────────────────────────────────────────

def add_indicators(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()

    # Taker buy ratio
    df["tbr"] = df["taker_buy_base"] / df["volume"].replace(0, np.nan)

    # MA20
    df["ma20"] = df["close"].rolling(REVERSAL_MA_PERIOD).mean()

    # 4h trend MA (48 x 5m candles)
    df["ma_trend"] = df["close"].rolling(TREND_MA_PERIOD).mean()

    # 4h decline: compare current close to close 48 candles ago (48 * 5m = 4h)
    df["decline_4h"] = df["close"].pct_change(48)

    # Bollinger Bands
    rolling = df["close"].rolling(BREAKOUT_BB_PERIOD)
    df["bb_mid"]   = rolling.mean()
    df["bb_std"]   = rolling.std()
    df["bb_upper"] = df["bb_mid"] + 2 * df["bb_std"]
    df["bb_width"] = (4 * df["bb_std"]) / df["bb_mid"]

    # BB width percentile over past 60 candles
    df["bb_width_pct"] = df["bb_width"].rolling(60).rank(pct=True) * 100

    # Volume MA20
    df["vol_ma20"] = df["volume"].rolling(20).mean()

    return df


# ── Signal detection ──────────────────────────────────────────────────────────

def detect_signals(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()

    # Previous 3 candle TBR max
    df["tbr_prev3_max"] = df["tbr"].shift(1).rolling(3).max()

    # Trend up filter: price above 4h MA
    trend_up = df["close"] > df["ma_trend"]

    # Reversal signal
    df["sig_reversal"] = (
        (df["tbr"] > REVERSAL_TAKER_BUY_THRESH) &
        (df["tbr_prev3_max"] < REVERSAL_PREV_TAKER_MAX) &
        (df["close"] < df["ma20"]) &
        (df["decline_4h"] < REVERSAL_4H_DECLINE) &
        trend_up
    )

    # Breakout signal
    df["sig_breakout"] = (
        (df["bb_width_pct"] < BREAKOUT_BB_SQUEEZE_PCT) &
        (df["volume"] > BREAKOUT_VOLUME_MULT * df["vol_ma20"]) &
        (df["tbr"] > BREAKOUT_TAKER_BUY_THRESH) &
        (df["close"] > df["bb_upper"]) &
        trend_up
    )

    df["signal"] = np.where(
        df["sig_reversal"], "reversal",
        np.where(df["sig_breakout"], "breakout", None)
    )

    return df


# ── Trade simulation ──────────────────────────────────────────────────────────

def simulate_trades(df: pd.DataFrame, symbol: str) -> list[dict]:
    trades = []
    in_trade = False
    entry_price = 0.0
    entry_time = None
    entry_signal = None
    hold_count = 0

    closes = df["close"].values
    signals = df["signal"].values
    times = df.index

    for i in range(1, len(df)):
        if in_trade:
            hold_count += 1
            pct = (closes[i] - entry_price) / entry_price

            hit_sl = pct <= STOP_LOSS
            hit_tp = pct >= TAKE_PROFIT
            hit_max = hold_count >= MAX_HOLD_CANDLES

            if hit_sl or hit_tp or hit_max:
                reason = "SL" if hit_sl else ("TP" if hit_tp else "MAX")
                trades.append({
                    "symbol": symbol,
                    "signal": entry_signal,
                    "entry_time": entry_time,
                    "exit_time": times[i],
                    "entry_price": entry_price,
                    "exit_price": closes[i],
                    "pct": pct,
                    "result": "win" if pct > 0 else "loss",
                    "reason": reason,
                    "month": entry_time.strftime("%Y-%m"),
                })
                in_trade = False

        # Enter on next candle after signal
        if not in_trade and i > 0 and signals[i - 1] is not None:
            in_trade = True
            entry_price = closes[i]  # next candle open ≈ prev close (5m)
            entry_time = times[i]
            entry_signal = signals[i - 1]
            hold_count = 0

    return trades


# ── Analysis ──────────────────────────────────────────────────────────────────

def analyze(trades: list[dict]) -> None:
    if not trades:
        print("No trades found.")
        return

    df = pd.DataFrame(trades)

    print("\n" + "=" * 70)
    print("OVERALL SUMMARY")
    print("=" * 70)

    for sig_type in ["reversal", "breakout", "all"]:
        sub = df if sig_type == "all" else df[df["signal"] == sig_type]
        if sub.empty:
            continue
        wins = (sub["result"] == "win").sum()
        total = len(sub)
        avg_pct = sub["pct"].mean() * 100
        total_pct = sub["pct"].sum() * 100
        print(f"\n[{sig_type.upper()}]  trades={total}  win_rate={wins/total:.1%}"
              f"  avg={avg_pct:.2f}%  total_return={total_pct:.1f}%")

    print("\n" + "=" * 70)
    print("MONTHLY BREAKDOWN (all signals combined)")
    print("=" * 70)

    monthly = (
        df.groupby("month")
        .agg(
            trades=("pct", "count"),
            wins=("result", lambda x: (x == "win").sum()),
            avg_pct=("pct", lambda x: x.mean() * 100),
            total_pct=("pct", lambda x: x.sum() * 100),
        )
        .reset_index()
    )
    monthly["win_rate"] = monthly["wins"] / monthly["trades"]
    monthly["avg_pct"] = monthly["avg_pct"].map("{:.2f}%".format)
    monthly["total_pct"] = monthly["total_pct"].map("{:.1f}%".format)
    monthly["win_rate"] = monthly["win_rate"].map("{:.1%}".format)
    print(tabulate(monthly, headers="keys", tablefmt="simple", showindex=False))

    print("\n" + "=" * 70)
    print("PER SYMBOL SUMMARY")
    print("=" * 70)

    by_sym = (
        df.groupby("symbol")
        .agg(
            trades=("pct", "count"),
            wins=("result", lambda x: (x == "win").sum()),
            avg_pct=("pct", lambda x: x.mean() * 100),
            total_pct=("pct", lambda x: x.sum() * 100),
        )
        .reset_index()
    )
    by_sym["win_rate"] = by_sym["wins"] / by_sym["trades"]
    by_sym["avg_pct"] = by_sym["avg_pct"].map("{:.2f}%".format)
    by_sym["total_pct"] = by_sym["total_pct"].map("{:.1f}%".format)
    by_sym["win_rate"] = by_sym["win_rate"].map("{:.1%}".format)
    print(tabulate(by_sym, headers="keys", tablefmt="simple", showindex=False))

    print("\n" + "=" * 70)
    print("EXIT REASON BREAKDOWN")
    print("=" * 70)
    print(df.groupby(["signal", "reason"]).size().to_string())


# ── Main ──────────────────────────────────────────────────────────────────────

def main():
    all_trades = []

    # ── Load BTC trend ────────────────────────────────────────────────────────
    print("Loading BTC trend data...")
    btc = fetch_klines(BTC_SYMBOL, START_TS, END_TS)
    if btc.empty:
        print("  BTC data failed, running without BTC filter")
        btc_trend = None
    else:
        btc["btc_ma"] = btc["close"].rolling(BTC_TREND_MA).mean()
        btc_trend = (btc["close"] > btc["btc_ma"]).rename("btc_uptrend")
        print(f"  BTC uptrend {btc_trend.mean():.1%} of the time")

    # ── Per-symbol backtest ───────────────────────────────────────────────────
    for symbol in SYMBOLS:
        df = fetch_klines(symbol, START_TS, END_TS)
        if df.empty:
            print(f"  → skipped")
            continue
        df = add_indicators(df)
        df = detect_signals(df)

        # Apply BTC trend filter
        if btc_trend is not None:
            btc_aligned = btc_trend.reindex(df.index, method="ffill")
            df.loc[~btc_aligned.fillna(False), "signal"] = None
            filtered = df["signal"].notna().sum()
        else:
            filtered = df["signal"].notna().sum()

        sig_count = df["signal"].notna().sum()
        print(f"  → {sig_count} signals after BTC filter (was {filtered})")

        trades = simulate_trades(df, symbol)
        print(f"  → {len(trades)} trades executed")
        all_trades.extend(trades)

    analyze(all_trades)


if __name__ == "__main__":
    main()