← Back
"""
Step 2: Sweep Z Entry × TP × SL (core R:R params)
Uses cached data from step1. NATR/CHOP/Vol at defaults.

Usage:
  cd /home/app/trading-bot-bybit
  python3 backtests/step2_sweep_z_tp_sl.py
"""

import json
import time
import numpy as np
from itertools import product

# ============================================================
# FIXED CONFIG
# ============================================================
ORDER_USD = 7.0
LEVERAGE = 3
VWAP_PERIOD = 50
TAKER_FEE = 0.00055
DAYS = 30

# Fixed filters (defaults from current bot)
NATR_MIN = 0.5
NATR_MAX = 3.0
CHOP_MIN = 45
COOLDOWN = 12  # bars

# ============================================================
# SWEEP GRID — only core params
# ============================================================
Z_ENTRIES = [1.5, 1.8, 2.0, 2.5, 3.0, 3.5, 4.0]
Z_MAXES = [0, 3.0, 4.0, 5.0, 6.0]
TP_PCTS = [0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 5.0]
SL_PCTS = [0.5, 1.0, 1.5, 2.0, 3.0, 5.0, 8.0, 10.0]


def simulate(z_scores, natr, chop, closes, highs, lows, z_entry, z_max, tp_pct, sl_pct):
    n = len(closes)
    deals = []
    in_trade = False
    side = None
    entry_price = 0
    entry_bar = 0
    cooldown_until = 0

    for i in range(VWAP_PERIOD, n):
        if in_trade:
            z = z_scores[i]
            closed = False
            close_price = 0
            reason = ""

            if side == "LONG":
                tp_p = entry_price * (1 + tp_pct / 100)
                sl_p = entry_price * (1 - sl_pct / 100)
                if lows[i] <= sl_p:
                    close_price = sl_p; reason = "SL"; closed = True
                elif highs[i] >= tp_p:
                    close_price = tp_p; reason = "TP"; closed = True
                elif z >= -0.3 and closes[i] > entry_price:
                    close_price = closes[i]; reason = "Z-TP"; closed = True
            else:
                tp_p = entry_price * (1 - tp_pct / 100)
                sl_p = entry_price * (1 + sl_pct / 100)
                if highs[i] >= sl_p:
                    close_price = sl_p; reason = "SL"; closed = True
                elif lows[i] <= tp_p:
                    close_price = tp_p; reason = "TP"; closed = True
                elif z <= 0.3 and closes[i] < entry_price:
                    close_price = closes[i]; reason = "Z-TP"; closed = True

            if not closed and (i - entry_bar) >= 36:
                close_price = closes[i]; reason = "TIME"; closed = True

            if closed:
                qty = (ORDER_USD * LEVERAGE) / entry_price
                pnl = qty * (close_price - entry_price) if side == "LONG" else qty * (entry_price - close_price)
                pnl -= qty * entry_price * TAKER_FEE + qty * close_price * TAKER_FEE
                deals.append({"pnl": pnl, "reason": reason, "bars": i - entry_bar})
                in_trade = False
                cooldown_until = i + COOLDOWN
            continue

        if i < cooldown_until:
            continue

        z = z_scores[i]
        if abs(z) <= z_entry:
            continue
        if z_max > 0 and abs(z) > z_max:
            continue
        if natr[i] < NATR_MIN or (NATR_MAX > 0 and natr[i] > NATR_MAX):
            continue
        if CHOP_MIN > 0 and chop[i] < CHOP_MIN:
            continue

        side = "LONG" if z < -z_entry else "SHORT"
        entry_price = closes[i]
        entry_bar = i
        in_trade = True

    if in_trade:
        qty = (ORDER_USD * LEVERAGE) / entry_price
        cp = closes[-1]
        pnl = qty * (cp - entry_price) if side == "LONG" else qty * (entry_price - cp)
        pnl -= qty * entry_price * TAKER_FEE + qty * cp * TAKER_FEE
        deals.append({"pnl": pnl, "reason": "END", "bars": n - 1 - entry_bar})

    return deals


def main():
    print("=" * 80)
    print("  STEP 2: Sweep Z Entry × Z Max × TP × SL")
    print("  Fixed: NATR 0.5-3.0, CHOP >= 45, CD 12 bars")
    print("=" * 80)

    # Load cached data
    print("\n  Loading cached data...")
    with open("/home/app/trading-bot-bybit/backtests/data_cache_30d.json") as f:
        cache = json.load(f)

    symbols = cache["symbols"]
    print(f"  {len(symbols)} symbols, fetched {cache['fetched_at']}")

    # Prep numpy arrays
    sym_data = {}
    for sym, d in symbols.items():
        sym_data[sym] = {
            "closes": np.array(d["closes"]),
            "highs": np.array(d["highs"]),
            "lows": np.array(d["lows"]),
            "z": np.array(d["z"]),
            "natr": np.array(d["natr"]),
            "chop": np.array(d["chop"]),
        }

    # Filter impossible combos
    combos = []
    for z_e, z_m, tp, sl in product(Z_ENTRIES, Z_MAXES, TP_PCTS, SL_PCTS):
        if z_m > 0 and z_e >= z_m:
            continue
        combos.append((z_e, z_m, tp, sl))

    print(f"  {len(combos)} valid combinations\n")

    results = []
    t0 = time.time()

    for ci, (z_e, z_m, tp, sl) in enumerate(combos):
        all_deals = []
        for sym, sd in sym_data.items():
            deals = simulate(sd["z"], sd["natr"], sd["chop"],
                           sd["closes"], sd["highs"], sd["lows"],
                           z_e, z_m, tp, sl)
            all_deals.extend(deals)

        if len(all_deals) < 5:
            continue

        total_pnl = sum(d["pnl"] for d in all_deals)
        wins = [d for d in all_deals if d["pnl"] > 0]
        losses = [d for d in all_deals if d["pnl"] <= 0]
        wr = len(wins) / len(all_deals) * 100
        gp = sum(d["pnl"] for d in wins) if wins else 0
        gl = abs(sum(d["pnl"] for d in losses)) if losses else 0.001
        pf = gp / gl

        tp_cnt = sum(1 for d in all_deals if d["reason"] == "TP")
        ztp_cnt = sum(1 for d in all_deals if d["reason"] == "Z-TP")
        sl_cnt = sum(1 for d in all_deals if d["reason"] == "SL")
        time_cnt = sum(1 for d in all_deals if d["reason"] == "TIME")

        results.append({
            "z_entry": z_e, "z_max": z_m, "tp_pct": tp, "sl_pct": sl,
            "deals": len(all_deals),
            "pnl": round(total_pnl, 2),
            "pnl_day": round(total_pnl / DAYS, 2),
            "wr": round(wr, 1),
            "pf": round(pf, 2),
            "avg_win": round(gp / len(wins), 3) if wins else 0,
            "avg_loss": round(gl / len(losses), 3) if losses else 0,
            "tp": tp_cnt, "ztp": ztp_cnt, "sl": sl_cnt, "time": time_cnt,
        })

        if (ci + 1) % 200 == 0:
            elapsed = time.time() - t0
            print(f"  ... {ci+1}/{len(combos)} ({elapsed:.0f}s)")

    elapsed = time.time() - t0
    profitable = [r for r in results if r["pnl"] > 0]
    print(f"\n  Done in {elapsed:.0f}s! {len(results)} valid, {len(profitable)} profitable")

    # ============================================================
    # DISPLAY
    # ============================================================
    header = f"{'#':>3} {'Z':>4} {'Zmax':>4} {'TP%':>5} {'SL%':>5} | {'Deals':>5} {'PnL':>8} {'$/d':>6} {'WR%':>5} {'PF':>6} {'AvgW':>7} {'AvgL':>7} | {'TP':>4} {'ZTP':>4} {'SL':>4} {'TIM':>4}"

    def prow(i, r):
        zm = r["z_max"] if r["z_max"] > 0 else 0
        print(f"{i:>3} {r['z_entry']:>4.1f} {zm:>4.1f} {r['tp_pct']:>5.2f} {r['sl_pct']:>5.1f} | {r['deals']:>5} ${r['pnl']:>+7.2f} ${r['pnl_day']:>+5.2f} {r['wr']:>5.1f} {r['pf']:>6.2f} ${r['avg_win']:>6.3f} ${r['avg_loss']:>6.3f} | {r['tp']:>4} {r['ztp']:>4} {r['sl']:>4} {r['time']:>4}")

    # TOP 25 by PnL (min 15 deals)
    by_pnl = sorted([r for r in results if r["deals"] >= 15], key=lambda x: x["pnl"], reverse=True)
    print(f"\n{'='*110}")
    print(f"  TOP 25 by PnL (min 15 deals, 30 days)")
    print(f"{'='*110}")
    print(header); print("-" * 110)
    for i, r in enumerate(by_pnl[:25]):
        prow(i+1, r)

    # TOP 25 by PF (min 15 deals)
    by_pf = sorted([r for r in results if r["deals"] >= 15 and r["pf"] < 100], key=lambda x: x["pf"], reverse=True)
    print(f"\n{'='*110}")
    print(f"  TOP 25 by Profit Factor (min 15 deals)")
    print(f"{'='*110}")
    print(header); print("-" * 110)
    for i, r in enumerate(by_pf[:25]):
        prow(i+1, r)

    # BALANCED: PF > 1.2, >= 30 deals, WR > 50%
    balanced = sorted([r for r in results if r["pf"] > 1.2 and r["deals"] >= 30 and r["wr"] > 50],
                      key=lambda x: x["pnl"], reverse=True)
    if balanced:
        print(f"\n{'='*110}")
        print(f"  BALANCED (PF > 1.2, >= 30 deals, WR > 50%)")
        print(f"{'='*110}")
        print(header); print("-" * 110)
        for i, r in enumerate(balanced[:25]):
            prow(i+1, r)

    # Save
    out = "/home/app/trading-bot-bybit/backtests/results_step2_z_tp_sl.json"
    with open(out, "w") as f:
        json.dump({
            "fixed": {"natr_min": NATR_MIN, "natr_max": NATR_MAX, "chop_min": CHOP_MIN, "cooldown": COOLDOWN},
            "grid": {"z_entry": Z_ENTRIES, "z_max": Z_MAXES, "tp_pct": TP_PCTS, "sl_pct": SL_PCTS},
            "total_combos": len(combos),
            "valid": len(results),
            "profitable": len(profitable),
            "results": sorted(results, key=lambda x: x["pnl"], reverse=True),
        }, f, indent=2)
    print(f"\n  Saved to {out}")


if __name__ == "__main__":
    main()

📜 Git History

dd32dfdchore: initial commit — version control setup5 weeks ago
Show last diff
Loading...