← Back
"""
Backtest: Rick's TradingView params
=====================================
Z=1.8, Zmax=2.5, TP=5%, SL=0.5%, NATR 0.75-2.0, CHOP>=55, no SO
Tested on SIREN: PF 2.416, WR 55.56%, +$42.28 (36 deals, 35 days)

Now running across ALL Bybit USDT perps to validate.

Usage:
  python3 backtests/backtest_rick_params.py
"""

import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import json
import time
import numpy as np
from datetime import datetime
from pybit.unified_trading import HTTP

# ============================================================
# RICK'S PARAMS (from TradingView screenshots)
# ============================================================
Z_ENTRY = 1.8
Z_MAX = 2.5
Z_TP = 0.3
TP_PCT = 3.0
SL_PCT = 1.0
COOLDOWN_BARS = 14
TIME_STOP_BARS = 36

# Filters
NATR_MIN = 0.75
NATR_MAX = 2.0
CHOP_MIN = 55

# Fixed
ORDER_USD = 7.0
LEVERAGE = 3
VWAP_PERIOD = 50
MAKER_FEE = 0.0002
TAKER_FEE = 0.00055
TIMEFRAME = "5"
DAYS = 30  # 30 days to match TV ~35 day range
MIN_VOLUME_24H = 20_000_000

BLACKLIST = {"BTCUSDT", "ETHUSDT", "TRXUSDT", "USDCUSDT", "BTCPERP", "ETHPERP"}

session = HTTP(testnet=False)


def get_symbols():
    resp = session.get_tickers(category="linear")
    if resp["retCode"] != 0:
        return []
    symbols = []
    for t in resp["result"]["list"]:
        sym = t["symbol"]
        if not sym.endswith("USDT") or sym in BLACKLIST:
            continue
        vol = float(t.get("turnover24h", 0))
        if vol >= MIN_VOLUME_24H:
            symbols.append({"symbol": sym, "volume_24h": vol})
    symbols.sort(key=lambda x: x["volume_24h"], reverse=True)
    return symbols[:60]


def fetch_klines(symbol, interval, days):
    all_klines = []
    bars_needed = days * 24 * 60 // int(interval)
    end_time = int(datetime.now().timestamp() * 1000)
    while len(all_klines) < bars_needed:
        try:
            resp = session.get_kline(category="linear", symbol=symbol,
                                     interval=interval, limit=1000, end=end_time)
            if resp["retCode"] != 0:
                break
            items = resp["result"]["list"]
            if not items:
                break
            for item in items:
                all_klines.append({
                    "ts": int(item[0]),
                    "h": float(item[2]), "l": float(item[3]),
                    "c": float(item[4]), "v": float(item[5]),
                })
            end_time = int(items[-1][0]) - 1
            if len(items) < 1000:
                break
        except:
            break
    all_klines.reverse()
    seen = set()
    unique = []
    for k in all_klines:
        if k["ts"] not in seen:
            seen.add(k["ts"])
            unique.append(k)
    return unique[-bars_needed:] if len(unique) > bars_needed else unique


def calc_indicators(closes, highs, lows, volumes):
    n = len(closes)

    # Z-VWAP
    z_scores = np.zeros(n)
    for i in range(VWAP_PERIOD, n):
        h = highs[i-VWAP_PERIOD:i]
        l = lows[i-VWAP_PERIOD:i]
        c = closes[i-VWAP_PERIOD:i]
        v = volumes[i-VWAP_PERIOD:i]
        tp = (h + l + c) / 3
        ctv = np.cumsum(tp * v)
        cv = np.cumsum(v)
        cv_s = np.where(cv == 0, 1, cv)
        vwap_arr = ctv / cv_s
        dev = c - vwap_arr
        std = np.std(dev)
        if std > 0:
            z_scores[i] = (closes[i] - vwap_arr[-1]) / std

    # Rolling NATR 14
    natr = np.zeros(n)
    for i in range(14, n):
        trs = []
        for j in range(i-13, i+1):
            tr = max(highs[j]-lows[j], abs(highs[j]-closes[j-1]), abs(lows[j]-closes[j-1]))
            trs.append(tr)
        natr[i] = (np.mean(trs) / closes[i]) * 100 if closes[i] > 0 else 0

    # CHOP 14
    chop = np.full(n, 50.0)
    for i in range(14, n):
        atr_sum = 0
        for j in range(i-13, i+1):
            tr = max(highs[j]-lows[j], abs(highs[j]-closes[j-1]), abs(lows[j]-closes[j-1]))
            atr_sum += tr
        hi = np.max(highs[i-13:i+1])
        lo = np.min(lows[i-13:i+1])
        rng = hi - lo
        if rng > 0:
            chop[i] = 100 * np.log10(atr_sum / rng) / np.log10(14)

    return z_scores, natr, chop


def simulate(z_scores, natr, chop, closes, highs, lows):
    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, reason, closed = sl_p, "SL", True
                elif highs[i] >= tp_p:
                    close_price, reason, closed = tp_p, "TP", True
                elif z >= -Z_TP and closes[i] > entry_price:
                    close_price, reason, closed = closes[i], "Z-TP", 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, reason, closed = sl_p, "SL", True
                elif lows[i] <= tp_p:
                    close_price, reason, closed = tp_p, "TP", True
                elif z <= Z_TP and closes[i] < entry_price:
                    close_price, reason, closed = closes[i], "Z-TP", True

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

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

        if i < cooldown_until:
            continue

        z = z_scores[i]
        if abs(z) <= Z_ENTRY:
            continue
        if abs(z) > Z_MAX:
            continue
        if natr[i] < NATR_MIN or natr[i] > NATR_MAX:
            continue
        if 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)
        fees = qty * entry_price * TAKER_FEE + qty * cp * TAKER_FEE
        pnl -= fees
        deals.append({"side": side, "pnl": pnl, "reason": "END", "bars": n-1-entry_bar,
                      "entry": entry_price, "exit": cp})

    return deals


def main():
    print("=" * 80)
    print("  RICK'S PARAMS BACKTEST")
    print(f"  Z={Z_ENTRY} Zmax={Z_MAX} TP={TP_PCT}% SL={SL_PCT}% CD={COOLDOWN_BARS}")
    print(f"  NATR {NATR_MIN}-{NATR_MAX}% | CHOP>={CHOP_MIN} | TimeStop={TIME_STOP_BARS} bars")
    print(f"  Period: {DAYS} days | SO=0 | Order=${ORDER_USD} x{LEVERAGE}")
    print("=" * 80)

    symbols_data = get_symbols()
    print(f"\n  {len(symbols_data)} symbols with vol >= ${MIN_VOLUME_24H/1e6:.0f}M")

    all_results = {}
    total_deals_all = []

    for idx, sd in enumerate(symbols_data):
        sym = sd["symbol"]
        print(f"  [{idx+1}/{len(symbols_data)}] {sym:15s} Vol ${sd['volume_24h']/1e6:.0f}M...", end=" ", flush=True)
        time.sleep(0.12)

        klines = fetch_klines(sym, TIMEFRAME, days=DAYS)
        if len(klines) < VWAP_PERIOD + 100:
            print(f"skip ({len(klines)} bars)")
            continue

        closes = np.array([k["c"] for k in klines])
        highs = np.array([k["h"] for k in klines])
        lows = np.array([k["l"] for k in klines])
        volumes = np.array([k["v"] for k in klines])

        z, natr, chop = calc_indicators(closes, highs, lows, volumes)
        deals = simulate(z, natr, chop, closes, highs, lows)

        for d in deals:
            d["symbol"] = sym
        total_deals_all.extend(deals)

        if deals:
            pnl = sum(d["pnl"] for d in deals)
            wins = sum(1 for d in deals if d["pnl"] > 0)
            wr = wins / len(deals) * 100
            gp = sum(d["pnl"] for d in deals if d["pnl"] > 0)
            gl = abs(sum(d["pnl"] for d in deals if d["pnl"] <= 0)) or 0.001
            pf = gp / gl
            emoji = "+" if pnl > 0 else "-"
            print(f"{len(deals):>3} deals  PnL ${pnl:>+7.2f}  WR {wr:.0f}%  PF {pf:.2f}")
            all_results[sym] = {"deals": len(deals), "pnl": round(pnl, 2), "wr": round(wr, 1), "pf": round(pf, 2)}
        else:
            print("0 deals")

    # ============================================================
    # AGGREGATE
    # ============================================================
    if not total_deals_all:
        print("\n  No deals!")
        return

    total_pnl = sum(d["pnl"] for d in total_deals_all)
    wins = [d for d in total_deals_all if d["pnl"] > 0]
    losses = [d for d in total_deals_all if d["pnl"] <= 0]
    wr = len(wins) / len(total_deals_all) * 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
    avg_bars = sum(d["bars"] for d in total_deals_all) / len(total_deals_all)

    longs = [d for d in total_deals_all if d["side"] == "LONG"]
    shorts = [d for d in total_deals_all if d["side"] == "SHORT"]
    long_pnl = sum(d["pnl"] for d in longs)
    short_pnl = sum(d["pnl"] for d in shorts)

    reasons = {}
    for d in total_deals_all:
        reasons[d["reason"]] = reasons.get(d["reason"], 0) + 1

    print(f"\n{'='*80}")
    print(f"  TOTAL RESULTS — {DAYS} days, {len(all_results)} symbols with trades")
    print(f"{'='*80}")
    e = "+" if total_pnl > 0 else "-"
    print(f"\n  {'$':>1} Total PnL:     ${total_pnl:+.2f}")
    print(f"  Profit Factor: {pf:.2f}")
    print(f"  Win Rate:      {wr:.1f}% ({len(wins)}W / {len(losses)}L)")
    print(f"  Avg Win:       ${gp/len(wins):.3f}" if wins else "")
    print(f"  Avg Loss:      ${gl/len(losses):.3f}" if losses else "")
    print(f"  Deals:         {len(total_deals_all)} (L:{len(longs)} S:{len(shorts)})")
    print(f"  Avg Duration:  {avg_bars:.1f} bars ({avg_bars*5/60:.1f}h)")
    print(f"  LONG PnL:      ${long_pnl:+.2f} ({len(longs)} deals)")
    print(f"  SHORT PnL:     ${short_pnl:+.2f} ({len(shorts)} deals)")

    print(f"\n  Close Reasons:")
    for r, cnt in sorted(reasons.items(), key=lambda x: -x[1]):
        print(f"    {r:10s} {cnt:>4d} ({cnt/len(total_deals_all)*100:.0f}%)")

    # Per-symbol sorted by PnL
    sorted_syms = sorted(all_results.items(), key=lambda x: x[1]["pnl"], reverse=True)
    print(f"\n  {'Symbol':15s} {'Deals':>5} {'PnL':>9} {'WR':>5} {'PF':>6}")
    print(f"  {'-'*45}")
    for sym, st in sorted_syms:
        e = "+" if st["pnl"] > 0 else "-"
        print(f"  {sym:15s} {st['deals']:>5} ${st['pnl']:>+7.2f} {st['wr']:>5.1f}% {st['pf']:>5.2f}")

    # Save
    output = os.path.join(os.path.dirname(__file__), "results_rick_params.json")
    with open(output, "w") as f:
        json.dump({
            "params": {"z_entry": Z_ENTRY, "z_max": Z_MAX, "z_tp": Z_TP,
                       "tp_pct": TP_PCT, "sl_pct": SL_PCT, "cooldown": COOLDOWN_BARS,
                       "natr_min": NATR_MIN, "natr_max": NATR_MAX, "chop_min": CHOP_MIN,
                       "time_stop": TIME_STOP_BARS, "order_usd": ORDER_USD, "leverage": LEVERAGE},
            "summary": {"total_pnl": round(total_pnl, 2), "deals": len(total_deals_all),
                        "wr": round(wr, 1), "pf": round(pf, 2)},
            "per_symbol": all_results,
        }, f, indent=2)
    print(f"\n  Saved to {output}")


if __name__ == "__main__":
    main()

📜 Git History

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