← Назад
""" 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()