โ ะะฐะทะฐะด"""
Session Grid Backtest โ Multi-Timeframe (15m, 1h)
===================================================
Simulates the ROTATION strategy:
1. Screener finds coin in RANGE (low ADX, narrow BB)
2. Grid runs within that range
3. Breakout detected โ close all, lock profit
4. Find next coin in range, repeat
Compare 15m vs 1h for screener/grid signals.
Usage:
python backtest_session_grid.py
"""
import ccxt
import pandas as pd
import numpy as np
import json
import time
from datetime import datetime, timedelta
# ============================================================
# CONFIG
# ============================================================
COINS = [
"ENA/USDT", "PENGU/USDT", "NEAR/USDT", "WLD/USDT", "UNI/USDT",
"VIRTUAL/USDT", "SOL/USDT", "XMR/USDT", "AVAX/USDT", "1000PEPE/USDT",
"DOT/USDT", "1000SHIB/USDT", "ICP/USDT", "ETH/USDT", "TON/USDT",
"LINK/USDT", "DOGE/USDT", "XRP/USDT", "AAVE/USDT", "ADA/USDT",
"FIL/USDT", "ONDO/USDT", "SUI/USDT", "OP/USDT", "ARB/USDT",
]
TIMEFRAMES = ["15m", "1h"]
LOOKBACK_DAYS = 30 # 30 days of data
# Grid params
GRID_SPACING_PCT = 0.1 # 0.1% between levels
GRID_LEVELS = 8 # 8 per side
POSITION_SIZE = 3.0 # $3 per level
LEVERAGE = 10
MAKER_FEE = 0.0002 # 0.02%
TAKER_FEE = 0.0004 # 0.04%
# Screener โ entry (range detection)
BB_PERIOD = 20
BB_STD = 2.0
BB_WIDTH_MIN = 0.3 # min BB width %
BB_WIDTH_MAX = 1.5 # max for ranging
ADX_MAX_ENTRY = 25 # ADX < 25 = range (stricter for entry)
ADX_PERIOD = 14
# Screener โ exit (breakout detection)
BREAKOUT_BB_MULT = 1.5 # BB expands 1.5x beyond max โ breakout
BREAKOUT_ADX = 30 # ADX > 30 โ confirmed trend
# Session constraints
MIN_SESSION_BARS = 10 # min bars before allowing exit
MAX_SESSION_BARS = {"15m": 192, "1h": 48} # ~2 days max per session
# Risk
MAX_LOSS_PCT = 3.0 # % of deposit, per session stop-loss
def fetch_klines(exchange, symbol, timeframe, days):
"""Fetch historical klines from Binance Futures."""
all_klines = []
since = exchange.parse8601((datetime.utcnow() - timedelta(days=days)).isoformat())
limit = 1000
while True:
try:
klines = exchange.fetch_ohlcv(symbol, timeframe, since=since, limit=limit)
except Exception as e:
print(f" Error fetching {symbol} {timeframe}: {e}")
return None
if not klines:
break
all_klines.extend(klines)
since = klines[-1][0] + 1
if len(klines) < limit:
break
time.sleep(0.15)
if not all_klines:
return None
df = pd.DataFrame(all_klines, columns=['ts', 'open', 'high', 'low', 'close', 'volume'])
df['ts'] = pd.to_datetime(df['ts'], unit='ms')
return df
def calc_bb_width(close, period=BB_PERIOD, std=BB_STD):
"""Bollinger Band width as % of mid."""
mid = close.rolling(period).mean()
s = close.rolling(period).std()
upper = mid + std * s
lower = mid - std * s
width = (upper - lower) / mid * 100
return width
def calc_adx(df, period=ADX_PERIOD):
"""ADX series."""
high = df['high']
low = df['low']
close = df['close']
plus_dm = high.diff()
minus_dm = -low.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)
tr1 = high - low
tr2 = (high - close.shift(1)).abs()
tr3 = (low - close.shift(1)).abs()
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
atr = tr.ewm(alpha=1/period, min_periods=period).mean()
plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr)
minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr)
dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di + 1e-10)
adx = dx.ewm(alpha=1/period, min_periods=period).mean()
return adx
def is_range_entry(bb_width, adx):
"""Check if conditions say 'range' โ start grid."""
return (BB_WIDTH_MIN <= bb_width <= BB_WIDTH_MAX) and (adx < ADX_MAX_ENTRY)
def is_breakout_exit(bb_width, adx):
"""Check if conditions say 'breakout' โ stop grid."""
return (bb_width > BB_WIDTH_MAX * BREAKOUT_BB_MULT) and (adx > BREAKOUT_ADX)
def simulate_grid_session(df_session, price_entry):
"""
Simulate grid within a session (range of bars).
Returns PnL from grid round-trips.
Simple model:
- Grid centered at price_entry
- Buy levels below, sell levels above (spacing 0.1%)
- Each time price crosses a level โ round-trip profit = spacing - fees
"""
closes = df_session['close'].values
highs = df_session['high'].values
lows = df_session['low'].values
spacing = price_entry * GRID_SPACING_PCT / 100
if spacing <= 0:
return 0, 0, 0
total_pnl = 0
round_trips = 0
max_drawdown = 0
running_pnl = 0
# Track grid fills using high/low range per bar
prev_level = int(closes[0] / spacing)
for i in range(1, len(closes)):
# How many levels did price traverse in this bar?
bar_low_level = int(lows[i] / spacing)
bar_high_level = int(highs[i] / spacing)
curr_level = int(closes[i] / spacing)
# Levels traversed = range of levels touched
levels_crossed = abs(curr_level - prev_level)
if levels_crossed > 0:
# Each crossing = one grid fill = partial round-trip
# A full round-trip = buy low + sell high = 2 fills
# PnL per RT = spacing * position_size / price - 2 * fee
rt_pnl_per_level = (POSITION_SIZE * LEVERAGE * GRID_SPACING_PCT / 100) - \
(POSITION_SIZE * LEVERAGE * MAKER_FEE * 2)
# Conservative: count round-trips as crossings / 2
rts = levels_crossed / 2
pnl = rts * rt_pnl_per_level
total_pnl += pnl
round_trips += rts
running_pnl += pnl
# Unrealized P&L from price drift (position exposure)
# Grid accumulates position in trend direction โ this is the risk
drift_from_center = (closes[i] - price_entry) / price_entry
# Net exposure grows as price moves from center
levels_from_center = abs(int(closes[i] / spacing) - int(price_entry / spacing))
exposure = min(levels_from_center, GRID_LEVELS) * POSITION_SIZE * LEVERAGE
unrealized = -abs(drift_from_center) * exposure # always negative (hedged but not fully)
current_total = running_pnl + unrealized
if current_total < max_drawdown:
max_drawdown = current_total
prev_level = curr_level
return round(total_pnl, 4), round(round_trips, 1), round(max_drawdown, 4)
def backtest_coin_tf(df, timeframe, deposit=50):
"""
Run session grid backtest on one coin/timeframe.
Returns list of sessions with PnL.
"""
bb_width = calc_bb_width(df['close'])
adx = calc_adx(df)
max_bars = MAX_SESSION_BARS.get(timeframe, 96)
sessions = []
i = BB_PERIOD + ADX_PERIOD # start after indicators warm up
in_session = False
session_start = 0
entry_price = 0
while i < len(df):
bw = bb_width.iloc[i]
ax = adx.iloc[i]
if pd.isna(bw) or pd.isna(ax):
i += 1
continue
if not in_session:
# Look for range entry
if is_range_entry(bw, ax):
in_session = True
session_start = i
entry_price = df['close'].iloc[i]
i += 1
continue
else:
# In session โ check for exit
bars_in = i - session_start
should_exit = False
exit_reason = ""
if bars_in >= MIN_SESSION_BARS and is_breakout_exit(bw, ax):
should_exit = True
exit_reason = "breakout"
elif bars_in >= max_bars:
should_exit = True
exit_reason = "max_time"
if should_exit:
# Simulate grid for this session
session_df = df.iloc[session_start:i+1]
pnl, rts, mdd = simulate_grid_session(session_df, entry_price)
# Apply session stop-loss
if mdd < -(deposit * MAX_LOSS_PCT / 100):
pnl = -(deposit * MAX_LOSS_PCT / 100)
exit_reason += "+stoploss"
# Close costs (market close = taker on remaining exposure)
close_cost = GRID_LEVELS * POSITION_SIZE * LEVERAGE * TAKER_FEE
pnl -= close_cost
sessions.append({
"start": str(df['ts'].iloc[session_start]),
"end": str(df['ts'].iloc[i]),
"bars": bars_in,
"entry_price": round(entry_price, 4),
"exit_price": round(df['close'].iloc[i], 4),
"pnl": round(pnl, 4),
"round_trips": rts,
"max_dd": round(mdd, 4),
"exit_reason": exit_reason,
"bb_width_exit": round(bw, 3),
"adx_exit": round(ax, 1),
})
in_session = False
# Cooldown: skip a few bars after exit
i += 3
continue
i += 1
return sessions
def main():
print("=" * 70)
print("SESSION GRID BACKTEST โ 15m vs 1h")
print(f"Period: {LOOKBACK_DAYS} days | Grid: {GRID_SPACING_PCT}% spacing, "
f"${POSITION_SIZE}ร{LEVERAGE}x per level")
print(f"Entry: BB {BB_WIDTH_MIN}-{BB_WIDTH_MAX}%, ADX<{ADX_MAX_ENTRY}")
print(f"Exit: BB>{BB_WIDTH_MAX*BREAKOUT_BB_MULT}% + ADX>{BREAKOUT_ADX}")
print("=" * 70)
exchange = ccxt.binanceusdm({
'enableRateLimit': True,
'options': {'defaultType': 'future'},
})
results = {}
for tf in TIMEFRAMES:
print(f"\n{'='*50}")
print(f" TIMEFRAME: {tf}")
print(f"{'='*50}")
tf_results = []
for symbol in COINS:
print(f"\n {symbol} ({tf})...", end=" ", flush=True)
df = fetch_klines(exchange, symbol, tf, LOOKBACK_DAYS)
if df is None or len(df) < BB_PERIOD + ADX_PERIOD + 20:
print("SKIP (no data)")
continue
sessions = backtest_coin_tf(df, tf)
if not sessions:
print(f"0 sessions")
continue
total_pnl = sum(s['pnl'] for s in sessions)
avg_pnl = total_pnl / len(sessions)
win_rate = sum(1 for s in sessions if s['pnl'] > 0) / len(sessions) * 100
avg_bars = np.mean([s['bars'] for s in sessions])
avg_rts = np.mean([s['round_trips'] for s in sessions])
total_rts = sum(s['round_trips'] for s in sessions)
print(f"{len(sessions)} sessions | PnL ${total_pnl:+.2f} | "
f"WR {win_rate:.0f}% | avg ${avg_pnl:+.2f}/session | "
f"avg {avg_bars:.0f} bars | {total_rts:.0f} RTs")
tf_results.append({
"symbol": symbol,
"sessions": len(sessions),
"total_pnl": round(total_pnl, 4),
"avg_pnl": round(avg_pnl, 4),
"win_rate": round(win_rate, 1),
"avg_bars": round(avg_bars, 1),
"avg_round_trips": round(avg_rts, 1),
"total_round_trips": round(total_rts, 1),
"max_dd": round(min(s['max_dd'] for s in sessions), 4),
"sessions_detail": sessions,
})
results[tf] = tf_results
# ============================================================
# SUMMARY
# ============================================================
print("\n\n" + "=" * 70)
print("SUMMARY โ SESSION GRID BACKTEST")
print("=" * 70)
for tf in TIMEFRAMES:
tf_data = results.get(tf, [])
if not tf_data:
print(f"\n{tf}: No data")
continue
total_pnl = sum(c['total_pnl'] for c in tf_data)
total_sessions = sum(c['sessions'] for c in tf_data)
total_rts = sum(c['total_round_trips'] for c in tf_data)
winners = [c for c in tf_data if c['total_pnl'] > 0]
losers = [c for c in tf_data if c['total_pnl'] <= 0]
print(f"\n๐ {tf} TIMEFRAME:")
print(f" Total PnL: ${total_pnl:+.2f}")
print(f" Sessions: {total_sessions} | Round-trips: {total_rts:.0f}")
print(f" Profitable coins: {len(winners)}/{len(tf_data)}")
if total_sessions > 0:
print(f" Avg PnL/session: ${total_pnl/total_sessions:+.2f}")
# Top 5 coins
sorted_coins = sorted(tf_data, key=lambda c: c['total_pnl'], reverse=True)
print(f"\n Top 5:")
for c in sorted_coins[:5]:
print(f" ๐ข {c['symbol']:15s} ${c['total_pnl']:+8.2f} "
f"({c['sessions']} sess, WR {c['win_rate']:.0f}%, "
f"{c['total_round_trips']:.0f} RTs)")
print(f"\n Bottom 5:")
for c in sorted_coins[-5:]:
print(f" ๐ด {c['symbol']:15s} ${c['total_pnl']:+8.2f} "
f"({c['sessions']} sess, WR {c['win_rate']:.0f}%)")
# Compare
print("\n\n" + "=" * 70)
print("15m vs 1h COMPARISON")
print("=" * 70)
for tf in TIMEFRAMES:
tf_data = results.get(tf, [])
total = sum(c['total_pnl'] for c in tf_data)
sess = sum(c['sessions'] for c in tf_data)
rts = sum(c['total_round_trips'] for c in tf_data)
avg = total / sess if sess > 0 else 0
print(f" {tf:4s}: ${total:+8.2f} total | {sess:3d} sessions | "
f"{rts:6.0f} RTs | ${avg:+.2f}/session")
# Save results
output_file = "/home/app/trading-bot/grid-bot/backtests/results_session_grid.json"
# Strip session details for compact JSON
compact = {}
for tf in TIMEFRAMES:
compact[tf] = []
for c in results.get(tf, []):
entry = {k: v for k, v in c.items() if k != 'sessions_detail'}
compact[tf].append(entry)
with open(output_file, 'w') as f:
json.dump(compact, f, indent=2)
print(f"\n๐พ Results saved: {output_file}")
if __name__ == "__main__":
main()