← Назад
""" Range Grid Backtest — растянутая сетка на весь боковик Стратегия Rick'а (классика): 1. Находим боковик на 1H (range 5-20%) 2. Растягиваем 20-40 гридов на весь диапазон 3. Крутим round-trips пока цена внутри 4. Когда пробой — фиксим всё, ищем другую монету 5. Мультимонетный — до 3 монет одновременно Тестируем варианты: A) 20 levels, range detection 48H lookback B) 30 levels, range detection 48H C) 40 levels, range detection 48H D) 30 levels, range detection 72H (шире диапазон) + Sideways filter score >= 45 для входа + Exit: price breaks range ±1% buffer Usage: python3 backtest_range_grid.py """ import requests import pandas as pd import numpy as np import time import json from datetime import datetime from pathlib import Path # ============================================================ # CONFIG # ============================================================ SYMBOLS = [ "ETHUSDT", "DOGEUSDT", "PENGUUSDT", "ENAUSDT", "NEARUSDT", "WLDUSDT", "SOLUSDT", "ARBUSDT", "XRPUSDT", "LINKUSDT", "SUIUSDT", "OPUSDT", "ADAUSDT", "UNIUSDT", "AVAXUSDT", ] DAYS_BACK = 30 # longer period — need time for ranges to form and play out LEVERAGE = 10 DEPOSIT = 50.0 FEE_PCT = 0.02 / 100 # maker MAX_LOSS_PCT = 5.0 # wider stop for range grid # Screener BB_PERIOD = 20 BB_STD = 2.0 ADX_PERIOD = 14 RANGE_LOOKBACK_1H = 24 # for sideways score # Range grid variants CONFIGS = [ {'name': 'A: 20lvl 48h', 'levels': 20, 'range_hours': 48, 'pos_usd': 2.5}, {'name': 'B: 30lvl 48h', 'levels': 30, 'range_hours': 48, 'pos_usd': 1.5}, {'name': 'C: 40lvl 48h', 'levels': 40, 'range_hours': 48, 'pos_usd': 1.0}, {'name': 'D: 30lvl 72h', 'levels': 30, 'range_hours': 72, 'pos_usd': 1.5}, {'name': 'E: 20lvl 72h', 'levels': 20, 'range_hours': 72, 'pos_usd': 2.5}, ] BREAKOUT_BUFFER_PCT = 1.0 # exit when price breaks range ± 1% MIN_RANGE_PCT = 4.0 # min range to consider (too tight = no profit) MAX_RANGE_PCT = 25.0 # max range (too wide = probably trending) MAX_SESSION_HOURS = 168 # max 7 days per session COOLDOWN_HOURS = 2 # gap between sessions per symbol # ============================================================ # DATA # ============================================================ def fetch_klines(symbol, interval, days_back): url = "https://fapi.binance.com/fapi/v1/klines" end_ts = int(time.time() * 1000) start_ts = int((time.time() - days_back * 86400) * 1000) all_candles = [] current_start = start_ts while current_start < end_ts: params = {"symbol": symbol, "interval": interval, "startTime": current_start, "limit": 1500} try: resp = requests.get(url, params=params, timeout=10) data = resp.json() if not isinstance(data, list) or len(data) == 0: break all_candles.extend(data) current_start = data[-1][0] + 1 time.sleep(0.08) except Exception as e: time.sleep(1) continue df = pd.DataFrame(all_candles, columns=[ 'timestamp', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_volume', 'trades', 'taker_buy_base', 'taker_buy_quote', 'ignore' ]) for col in ['open', 'high', 'low', 'close', 'volume', 'quote_volume']: df[col] = df[col].astype(float) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') df = df.drop_duplicates(subset='timestamp').sort_values('timestamp').reset_index(drop=True) return df # ============================================================ # SIDEWAYS SCREENER (1H) # ============================================================ def calc_screener_1h(df): df['bb_mid'] = df['close'].rolling(BB_PERIOD).mean() df['bb_std'] = df['close'].rolling(BB_PERIOD).std() df['bb_upper'] = df['bb_mid'] + BB_STD * df['bb_std'] df['bb_lower'] = df['bb_mid'] - BB_STD * df['bb_std'] df['bb_width'] = ((df['bb_upper'] - df['bb_lower']) / df['bb_mid']) * 100 h, l, c = df['high'], df['low'], df['close'] plus_dm = h.diff() minus_dm = -l.diff() plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0) minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0.0) tr = pd.concat([h - l, (h - c.shift(1)).abs(), (l - c.shift(1)).abs()], axis=1).max(axis=1) atr = tr.ewm(alpha=1/ADX_PERIOD, min_periods=ADX_PERIOD).mean() plus_di = 100 * (plus_dm.ewm(alpha=1/ADX_PERIOD, min_periods=ADX_PERIOD).mean() / atr) minus_di = 100 * (minus_dm.ewm(alpha=1/ADX_PERIOD, min_periods=ADX_PERIOD).mean() / atr) dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di + 1e-10) df['adx'] = dx.ewm(alpha=1/ADX_PERIOD, min_periods=ADX_PERIOD).mean() df['natr'] = (atr / c) * 100 df['range_high'] = df['high'].rolling(RANGE_LOOKBACK_1H).max() df['range_low'] = df['low'].rolling(RANGE_LOOKBACK_1H).min() rng = df['range_high'] - df['range_low'] df['range_pos'] = (df['close'] - df['range_low']) / rng.replace(0, np.nan) df['range_pct'] = (rng / df['close']) * 100 df['candle_dir'] = np.where(df['close'] > df['open'], 1, -1) df['dir_change'] = (df['candle_dir'] != df['candle_dir'].shift(1)).astype(int) df['dir_changes'] = df['dir_change'].rolling(RANGE_LOOKBACK_1H).sum() return df def sideways_score(row): score = 0 adx = row.get('adx', np.nan) bb_w = row.get('bb_width', np.nan) rp = row.get('range_pos', np.nan) dc = row.get('dir_changes', np.nan) natr = row.get('natr', np.nan) if any(pd.isna(x) for x in [adx, bb_w, rp, dc, natr]): return 0 if adx <= 5: score += 25 elif adx <= 20: score += 25 * (1 - (adx - 5) / 15) elif adx <= 30: score += max(0, -5 * (adx - 20) / 10) else: score -= 10 if 1.5 <= bb_w <= 4.0: score += 20 * (bb_w - 1.5) / 1.0 if bb_w <= 2.5 else 20 * (4.0 - bb_w) / 1.5 elif 0.5 <= bb_w < 1.5: score += 5 if 0.3 <= rp <= 0.7: score += 20 * (1 - abs(rp - 0.5) / 0.2) elif 0.2 <= rp < 0.3 or 0.7 < rp <= 0.8: score += 5 max_ch = RANGE_LOOKBACK_1H * 0.7 if dc >= 8: score += min(20, 20 * (dc - 8) / (max_ch - 8)) if 0.15 <= natr <= 0.6: score += 15 * (natr - 0.15) / 0.15 if natr <= 0.3 else 15 * (0.6 - natr) / 0.3 elif 0.1 <= natr < 0.15: score += 3 return max(0, round(score, 1)) # ============================================================ # RANGE GRID ENGINE — runs on 1H candles # ============================================================ def run_range_grid(df_1h, start_idx, n_levels, range_hours, pos_usd): """ 1. Look back range_hours to find high/low 2. Set grid levels across that range 3. Simulate on 1H candles until breakout or timeout """ lookback = range_hours # 1H candles = hours if start_idx < lookback + 5 or start_idx + 5 >= len(df_1h): return None # Detect range from lookback window = df_1h.iloc[start_idx - lookback:start_idx] range_high = window['high'].max() range_low = window['low'].min() range_pct = ((range_high - range_low) / range_low) * 100 if range_pct < MIN_RANGE_PCT or range_pct > MAX_RANGE_PCT: return None current_price = df_1h['close'].iloc[start_idx] # Check price is within range (not already broken out) if current_price > range_high or current_price < range_low: return None # Grid levels spread across entire range step = (range_high - range_low) / (n_levels + 1) grid_levels = [range_low + step * (k + 1) for k in range(n_levels)] # Split into buy levels (below price) and sell levels (above price) buy_levels = sorted([lv for lv in grid_levels if lv < current_price]) sell_levels = sorted([lv for lv in grid_levels if lv >= current_price]) # Track state buy_filled = {lv: False for lv in buy_levels} sell_filled = {lv: False for lv in sell_levels} positions = [] # (side, entry_price) pnl = 0.0 fees = 0.0 trades = 0 round_trips = 0 close_reason = 'timeout' # Breakout thresholds break_high = range_high * (1 + BREAKOUT_BUFFER_PCT / 100) break_low = range_low * (1 - BREAKOUT_BUFFER_PCT / 100) max_candles = min(MAX_SESSION_HOURS, len(df_1h) - start_idx - 1) for j in range(start_idx + 1, start_idx + max_candles): price = df_1h['close'].iloc[j] lo = df_1h['low'].iloc[j] hi = df_1h['high'].iloc[j] # Check buy fills (price dipped to level) for lv in buy_levels: if not buy_filled[lv] and lo <= lv: buy_filled[lv] = True positions.append(('long', lv)) fees += pos_usd * LEVERAGE * FEE_PCT trades += 1 # Check sell fills (price rose to level) for lv in sell_levels: if not sell_filled[lv] and hi >= lv: sell_filled[lv] = True positions.append(('short', lv)) fees += pos_usd * LEVERAGE * FEE_PCT trades += 1 # Round-trip matching: # Pair closest buy with closest sell that's above it # Sort positions longs = sorted([p for p in positions if p[0] == 'long'], key=lambda x: x[1]) shorts = sorted([p for p in positions if p[0] == 'short'], key=lambda x: x[1]) matched_longs = set() matched_shorts = set() for li, (_, buy_px) in enumerate(longs): for si, (_, sell_px) in enumerate(shorts): if si in matched_shorts: continue if sell_px > buy_px: # Round trip! qty = (pos_usd * LEVERAGE) / buy_px spread = sell_px - buy_px pnl += qty * spread fees += pos_usd * LEVERAGE * FEE_PCT * 2 # close fees round_trips += 1 matched_longs.add(li) matched_shorts.add(si) # Reset these levels so they can fill again buy_filled[buy_px] = False sell_filled[sell_px] = False break # Remove matched positions new_positions = [] for idx, p in enumerate(longs): if idx not in matched_longs: new_positions.append(p) for idx, p in enumerate(shorts): if idx not in matched_shorts: new_positions.append(p) positions = new_positions # BREAKOUT CHECK — exit all if price > break_high or price < break_low: # Close all open positions at current price for side, entry_px in positions: qty = (pos_usd * LEVERAGE) / entry_px if side == 'long': pnl += qty * (price - entry_px) else: pnl += qty * (entry_px - price) fees += pos_usd * LEVERAGE * FEE_PCT positions = [] if price > break_high: close_reason = 'breakout_up' else: close_reason = 'breakout_down' break # Max loss check unrealized = 0 for side, entry_px in positions: qty = (pos_usd * LEVERAGE) / entry_px if side == 'long': unrealized += qty * (price - entry_px) else: unrealized += qty * (entry_px - price) total_capital = n_levels * pos_usd if pnl + unrealized - fees < -(total_capital * MAX_LOSS_PCT / 100): for side, entry_px in positions: qty = (pos_usd * LEVERAGE) / entry_px if side == 'long': pnl += qty * (price - entry_px) else: pnl += qty * (entry_px - price) fees += pos_usd * LEVERAGE * FEE_PCT positions = [] close_reason = 'max_loss' break # Close remaining at end if positions: price = df_1h['close'].iloc[min(start_idx + max_candles - 1, len(df_1h) - 1)] for side, entry_px in positions: qty = (pos_usd * LEVERAGE) / entry_px if side == 'long': pnl += qty * (price - entry_px) else: pnl += qty * (entry_px - price) fees += pos_usd * LEVERAGE * FEE_PCT net = pnl - fees duration_h = j - start_idx if 'j' in dir() else 0 if trades == 0: return None return { 'pnl': round(net, 4), 'gross_pnl': round(pnl, 4), 'trades': trades, 'round_trips': round_trips, 'fees': round(fees, 4), 'close_reason': close_reason, 'range_pct': round(range_pct, 2), 'duration_h': duration_h, 'levels_used': n_levels, 'spacing_pct': round(range_pct / (n_levels + 1), 3), } # ============================================================ # MULTI-COIN ROTATION SIMULATOR # ============================================================ def simulate_rotation(all_sessions_by_symbol, max_concurrent=3): """ Simulate multi-coin rotation: - Sort all potential sessions by start time - Allow max N concurrent grids - When one exits, start next available """ # Flatten and sort by start time all_entries = [] for sym, sessions in all_sessions_by_symbol.items(): for s in sessions: all_entries.append({**s, 'symbol': sym}) all_entries.sort(key=lambda x: x['start_idx']) active = [] # (end_idx, session_data) completed = [] for entry in all_entries: # Remove finished sessions active = [(end, s) for end, s in active if end > entry['start_idx']] if len(active) < max_concurrent: end_idx = entry['start_idx'] + entry['duration_h'] active.append((end_idx, entry)) completed.append(entry) return completed # ============================================================ # MAIN # ============================================================ if __name__ == "__main__": print("=" * 75) print(" RANGE GRID BACKTEST — full range × 20-40 levels") print(f" {len(SYMBOLS)} coins | {DAYS_BACK}d | ${DEPOSIT} × {LEVERAGE}x") print("=" * 75) # Store results config_results = {c['name']: {'all': [], 'filtered': [], 'by_symbol': {}} for c in CONFIGS} for sym in SYMBOLS: print(f"\n 📊 {sym}...", end=" ", flush=True) df_1h = fetch_klines(sym, '1h', DAYS_BACK) if len(df_1h) < 100: print("skip") continue # Calc screener df_1h = calc_screener_1h(df_1h) df_1h['sw_score'] = df_1h.apply(sideways_score, axis=1) for cfg in CONFIGS: n_levels = cfg['levels'] range_hours = cfg['range_hours'] pos_usd = cfg['pos_usd'] cname = cfg['name'] # Scan every 4 hours for entry opportunities i = max(range_hours + 10, 50) sessions = [] cooldown_until = 0 while i < len(df_1h) - 5: if i < cooldown_until: i += 1 continue score = df_1h['sw_score'].iloc[i] adx = df_1h['adx'].iloc[i] range_pos = df_1h['range_pos'].iloc[i] if not pd.isna(df_1h['range_pos'].iloc[i]) else 0.5 result = run_range_grid(df_1h, i, n_levels, range_hours, pos_usd) if result: result['symbol'] = sym result['sw_score'] = round(score, 1) result['adx'] = round(adx, 1) if not pd.isna(adx) else None result['range_pos'] = round(range_pos, 2) result['start_idx'] = i result['ts'] = str(df_1h['timestamp'].iloc[i]) config_results[cname]['all'].append(result) if sym not in config_results[cname]['by_symbol']: config_results[cname]['by_symbol'][sym] = [] config_results[cname]['by_symbol'][sym].append(result) if score >= 40: config_results[cname]['filtered'].append(result) # Skip ahead by session duration (don't overlap) cooldown_until = i + max(result['duration_h'], COOLDOWN_HOURS) i += 4 # check every 4 hours print(f"done") # ============================================================ # REPORTS # ============================================================ print("\n" + "=" * 75) print(" 🏆 RESULTS") print("=" * 75) print(f"\n {'Config':<20} {'Mode':<6} {'Sess':>5} {'WR':>6} {'PnL':>10} {'Avg':>9} {'$/wk':>7} {'RTs':>5} {'AvgRng':>7} {'AvgDur':>7} {'ML':>4}") print(f" {'─'*90}") best_name = None best_avg = -999 for cfg in CONFIGS: cname = cfg['name'] for mode, label in [('all', 'ALL'), ('filtered', 'FILT')]: sess = config_results[cname][mode] if not sess: continue total_pnl = sum(s['pnl'] for s in sess) wins = len([s for s in sess if s['pnl'] > 0]) wr = 100 * wins / len(sess) avg = total_pnl / len(sess) rts = sum(s['round_trips'] for s in sess) ml = len([s for s in sess if s['close_reason'] == 'max_loss']) avg_range = np.mean([s['range_pct'] for s in sess]) avg_dur = np.mean([s['duration_h'] for s in sess]) per_week = total_pnl / DAYS_BACK * 7 emoji = '🟢' if total_pnl > 0 else '🔴' print(f" {emoji} {cname:<18} {label:<6} {len(sess):>5} {wr:>5.0f}% ${total_pnl:>8.2f} ${avg:>7.3f} ${per_week:>5.2f} {rts:>5} {avg_range:>6.1f}% {avg_dur:>5.0f}h {ml:>4}") if label == 'FILT' and avg > best_avg: best_avg = avg best_name = cname # Best config deep-dive if best_name: sess = config_results[best_name]['filtered'] print(f"\n{'='*75}") print(f" 🥇 ЛУЧШИЙ (filtered): {best_name}") print(f"{'='*75}") total_pnl = sum(s['pnl'] for s in sess) wins = [s for s in sess if s['pnl'] > 0] losses = [s for s in sess if s['pnl'] <= 0] print(f" Sessions: {len(sess)} | WR: {100*len(wins)/len(sess):.0f}%") print(f" Total PnL: ${total_pnl:.2f} | Per week: ${total_pnl/DAYS_BACK*7:.2f}") print(f" Per month (est): ${total_pnl/DAYS_BACK*30:.2f}") if wins: print(f" Avg win: ${sum(s['pnl'] for s in wins)/len(wins):.3f} | Best: ${max(s['pnl'] for s in wins):.3f}") if losses: print(f" Avg loss: ${sum(s['pnl'] for s in losses)/len(losses):.3f} | Worst: ${min(s['pnl'] for s in losses):.3f}") gross_w = sum(s['pnl'] for s in wins) if wins else 0 gross_l = abs(sum(s['pnl'] for s in losses)) if losses else 1 print(f" Profit factor: {gross_w/gross_l:.2f}") print(f" Avg range: {np.mean([s['range_pct'] for s in sess]):.1f}%") print(f" Avg duration: {np.mean([s['duration_h'] for s in sess]):.0f}h") print(f" Avg spacing: {np.mean([s['spacing_pct'] for s in sess]):.3f}%") print(f" Total round-trips: {sum(s['round_trips'] for s in sess)}") # Close reasons reasons = {} for s in sess: r = s['close_reason'] reasons[r] = reasons.get(r, 0) + 1 print(f" Close reasons: {reasons}") # Equity curve equity = [0] for s in sess: equity.append(equity[-1] + s['pnl']) eq = np.array(equity) peak = np.maximum.accumulate(eq) dd = eq - peak print(f" Max drawdown: ${dd.min():.3f}") # By range size print(f"\n PnL by range size:") print(f" {'Range':<12} {'Sess':>5} {'WR':>6} {'PnL':>10} {'Avg':>9} {'RTs':>5} {'AvgDur':>7}") print(f" {'─'*56}") for lo, hi in [(4, 7), (7, 10), (10, 15), (15, 20), (20, 25)]: sub = [s for s in sess if lo <= s['range_pct'] < hi] if not sub: continue tp = sum(s['pnl'] for s in sub) w = len([s for s in sub if s['pnl'] > 0]) wr = 100 * w / len(sub) rt = sum(s['round_trips'] for s in sub) ad = np.mean([s['duration_h'] for s in sub]) e = '🟢' if tp > 0 else '🔴' print(f" {e} {lo}-{hi}%{'':<7} {len(sub):>5} {wr:>5.0f}% ${tp:>8.2f} ${tp/len(sub):>7.3f} {rt:>5} {ad:>5.0f}h") # Per symbol print(f"\n Per symbol:") print(f" {'Symbol':<12} {'Sess':>5} {'WR':>6} {'PnL':>10} {'RTs':>5} {'AvgRng':>7}") print(f" {'─'*45}") syms = sorted(set(s['symbol'] for s in sess)) for sym in syms: sub = [s for s in sess if s['symbol'] == sym] tp = sum(s['pnl'] for s in sub) w = len([s for s in sub if s['pnl'] > 0]) wr = 100 * w / len(sub) rt = sum(s['round_trips'] for s in sub) ar = np.mean([s['range_pct'] for s in sub]) e = '🟢' if tp > 0 else '🔴' print(f" {e} {sym:<10} {len(sub):>5} {wr:>5.0f}% ${tp:>8.2f} {rt:>5} {ar:>6.1f}%") # By close reason detail print(f"\n PnL by close reason:") for reason in sorted(reasons.keys()): sub = [s for s in sess if s['close_reason'] == reason] tp = sum(s['pnl'] for s in sub) avg = tp / len(sub) if sub else 0 print(f" {reason:<20} {len(sub):>4} sess | ${tp:>8.3f} total | ${avg:>7.3f} avg") # Multi-coin rotation simulation (best config) if best_name: print(f"\n{'='*75}") print(f" 🔄 MULTI-COIN ROTATION (max 3 concurrent)") print(f"{'='*75}") by_sym = config_results[best_name]['by_symbol'] # Filter each symbol's sessions filtered_by_sym = {} for sym, sessions in by_sym.items(): filtered_by_sym[sym] = [s for s in sessions if s.get('sw_score', 0) >= 40] rotated = simulate_rotation(filtered_by_sym, max_concurrent=3) if rotated: tp = sum(s['pnl'] for s in rotated) w = len([s for s in rotated if s['pnl'] > 0]) wr = 100 * w / len(rotated) rt = sum(s['round_trips'] for s in rotated) print(f" Sessions taken: {len(rotated)}") print(f" WR: {wr:.0f}% | Total PnL: ${tp:.2f} | Per week: ${tp/DAYS_BACK*7:.2f}") print(f" Round-trips: {rt}") coins_used = set(s['symbol'] for s in rotated) print(f" Coins rotated: {len(coins_used)} ({', '.join(sorted(coins_used))})") # Top 10 sessions if best_name: sess = config_results[best_name]['filtered'] print(f"\n 📈 Top 10 sessions:") sorted_s = sorted(sess, key=lambda x: x['pnl'], reverse=True)[:10] for s in sorted_s: e = '🟢' if s['pnl'] > 0 else '🔴' print(f" {e} {s['symbol']:<10} {s['ts'][:13]} | rng {s['range_pct']}% | {s['duration_h']}h | {s['round_trips']}rt | ${s['pnl']:.3f} | {s['close_reason']}") print(f"\n 📉 Worst 5:") sorted_s = sorted(sess, key=lambda x: x['pnl'])[:5] for s in sorted_s: print(f" 🔴 {s['symbol']:<10} {s['ts'][:13]} | rng {s['range_pct']}% | {s['duration_h']}h | {s['round_trips']}rt | ${s['pnl']:.3f} | {s['close_reason']}") # Save output = { 'config': { 'symbols': SYMBOLS, 'days_back': DAYS_BACK, 'leverage': LEVERAGE, 'fee_pct': FEE_PCT * 100, 'breakout_buffer': BREAKOUT_BUFFER_PCT, 'min_range': MIN_RANGE_PCT, 'max_range': MAX_RANGE_PCT, }, 'configs_tested': CONFIGS, 'results': {name: { 'all_count': len(config_results[name]['all']), 'all_pnl': round(sum(s['pnl'] for s in config_results[name]['all']), 4) if config_results[name]['all'] else 0, 'filtered_count': len(config_results[name]['filtered']), 'filtered_pnl': round(sum(s['pnl'] for s in config_results[name]['filtered']), 4) if config_results[name]['filtered'] else 0, } for name in config_results}, 'best': best_name, 'tested_at': datetime.now().isoformat(), } out_path = Path(__file__).parent / 'results_range_grid.json' with open(out_path, 'w') as f: json.dump(output, f, indent=2) print(f"\n💾 {out_path}")