""" Grid engine — gioco "Grid Traders" (sessione 3), regola: STRATEGIA_GRIGLIA.md. 100 agenti ciechi ricevono due serie anonime (X=A=BTC, Y=B=ETH, mai rivelate) e propongono la CONFIGURAZIONE di una griglia di trading secondo la spec del documento STRATEGIA_GRIGLIA.md. Questo motore la backtesta in modo deterministico, causale e fee-aware: - griglia GEOMETRICA dentro un range definito al deploy su close[i] (§3.2): ratio = (RANGE_HIGH/RANGE_LOW)^(1/GRID_LEVELS), livello[k] = RL * ratio^k Il range e' parametrizzato in PERCENTUALE attorno al prezzo di deploy (range_down/range_up), cosi' la griglia e' backtestabile su tutta la storia. - capitale suddiviso in anticipo: quote_per_livello = 1/GRID_LEVELS (§3.3) - VINCOLO BREAK-EVEN (§4): passo > MARGINE(1.5) x costo round-trip. Se violato il motore SI RIFIUTA DI PARTIRE (come da spec): spec squalificata. - ciclo (§5.2): compra quote_per_livello su attraversamento VERSO IL BASSO di un livello non riempito; vendi quel livello su attraversamento VERSO L'ALTO del livello successivo. Livelli FISSI per tutto l'episodio (non inseguono il prezzo). - guardie (§5.2/§6): STOP-LOSS sotto RANGE_LOW e TAKE-PROFIT sopra RANGE_HIGH hanno priorita' su tutto: liquidano l'intera posizione e fermano la griglia. - episodi: quando una griglia muore (SL / TP / max_bars) se ne deploya una nuova sul prezzo corrente (il "riavvio del bot" di §6.6, qui automatizzato). Causalita': il deploy a close[i] usa solo close[i]; i fill avvengono dalle barre successive lungo il percorso intrabar O->L->H->C (se close>=open) o O->H->L->C. Fee 0.10% round-trip per livello (baseline Deribit del progetto) + slippage opzionale per lato (GAME_SLIP), come negli altri giochi. """ from __future__ import annotations import math from bisect import bisect_left, bisect_right import numpy as np from scripts.games.engine import load_anon, splits3, TF_BPM, FEE_RT MIN_TRADES_PER_MONTH = 10.0 MARGIN = 1.5 # margine di sicurezza del vincolo break-even (§4) _SLIP = 0.0 # slippage per LATO (oltre alle fee), come engine.py def set_slippage(slip_per_side: float): global _SLIP _SLIP = float(slip_per_side) def cost_rt(fee: float = FEE_RT) -> float: """Costo di un round-trip completo (fee RT + 2 lati di slippage).""" return fee + 2 * _SLIP def grid_ratio(p) -> float: """Ratio geometrico della griglia: indipendente dal prezzo di deploy.""" rd, ru, L = float(p["range_down"]), float(p["range_up"]), int(p["levels"]) return ((1.0 + ru) / (1.0 - rd)) ** (1.0 / L) def max_levels(range_down: float, range_up: float, fee: float = FEE_RT) -> int: """Massimo numero di livelli che rispetta il vincolo break-even (§4).""" width = math.log((1.0 + range_up) / (1.0 - range_down)) min_step = math.log(1.0 + MARGIN * cost_rt(fee)) return max(0, int(math.floor(width / min_step))) # -------------------------------------------------------------------------- # Backtest della griglia (episodi deploy -> SL/TP/timeout -> redeploy) # -------------------------------------------------------------------------- def _backtest_grid(o, p, fee=FEE_RT): """Ritorna l'array dei net-return per trade (round-trip o liquidazione), in frazione del capitale dell'episodio. None se il vincolo break-even e' violato (il bot si rifiuta di partire, §4).""" op, hi, lo, cl = o["open"], o["high"], o["low"], o["close"] n = len(cl) crt = cost_rt(fee) L = int(p["levels"]) rd, ru = float(p["range_down"]), float(p["range_up"]) slb, tpb = float(p["sl_buf"]), float(p["tp_buf"]) max_bars = max(1, int(p["max_bars"])) if L < 2: return None ratio = grid_ratio(p) step = ratio - 1.0 if step <= MARGIN * crt: return None # §4: vincolo break-even violato lstep = math.log(ratio) with np.errstate(divide="ignore"): llo = np.log(lo) lhi = np.log(hi) qpl = 1.0 / L rets = [] i = 20 # warmup minimo (parita' con engine.py) while i < n - 1: px = float(cl[i]) if not np.isfinite(px) or px <= 0: i += 1 continue rl_ = px * (1.0 - rd) lv = [rl_ * ratio ** k for k in range(L + 1)] # lv[L] = RANGE_HIGH sl = rl_ * (1.0 - slb) tp = lv[L] * (1.0 + tpb) off = math.log(rl_) end = min(n - 1, i + max_bars) # indice-cella (floor) di low/high per il fast-skip delle barre quiete klo = np.floor((llo[i + 1:end + 1] - off) / lstep).astype(np.int64) khi = np.floor((lhi[i + 1:end + 1] - off) / lstep).astype(np.int64) slhit = lo[i + 1:end + 1] <= sl tphit = hi[i + 1:end + 1] >= tp filled = [False] * L n_open = 0 cur = px kc = bisect_right(lv, cur) - 1 done = False exit_i = end for j in range(i + 1, end + 1): jj = j - (i + 1) if klo[jj] == khi[jj] == kc and not slhit[jj] and not tphit[jj]: cur = cl[j] # barra quieta: nessun livello toccato continue pts = (op[j], lo[j], hi[j], cl[j]) if cl[j] >= op[j] \ else (op[j], hi[j], lo[j], cl[j]) for q in pts: q = float(q) if q == cur: continue if q < cur: # discesa: fill dei buy-level attraversati (alto -> basso) k1 = bisect_left(lv, q) # primo livello >= q k2 = bisect_left(lv, cur) - 1 # ultimo livello < cur for k in range(min(k2, L - 1), max(k1, 0) - 1, -1): if not filled[k]: filled[k] = True n_open += 1 if q <= sl: # STOP-LOSS: vendi tutta la posizione a sl, ferma la griglia if n_open: rets.append(sum( qpl * (sl / lv[k] - 1.0 - crt) for k in range(L) if filled[k])) done = True cur = q break else: # salita: vendi i livelli riempiti il cui target e' attraversato m1 = bisect_right(lv, cur) # primo livello > cur m2 = bisect_right(lv, q) - 1 # ultimo livello <= q for m in range(max(m1, 1), min(m2, L) + 1): k = m - 1 if filled[k]: rets.append(qpl * (lv[m] / lv[k] - 1.0 - crt)) filled[k] = False n_open -= 1 if q >= tp: # TAKE-PROFIT: chiudi il residuo a tp, ferma la griglia if n_open: rets.append(sum( qpl * (tp / lv[k] - 1.0 - crt) for k in range(L) if filled[k])) done = True cur = q break cur = q if done: exit_i = j break kc = bisect_right(lv, cur) - 1 if not done: # timeout max_bars: liquida il residuo al close dell'ultima barra if n_open: rets.append(sum( qpl * (cl[end] / lv[k] - 1.0 - crt) for k in range(L) if filled[k])) exit_i = end i = exit_i # redeploy sul prezzo dove e' morta la griglia return np.array(rets) if rets else np.array([]) # -------------------------------------------------------------------------- # Valutazione + scoring (stessa fitness degli altri giochi) # -------------------------------------------------------------------------- def evaluate(data, spec, sl=None, fee=FEE_RT): """spec = {series: 'A'|'B', tf, params{range_down,range_up,levels,sl_buf, tp_buf,max_bars}}. Ritorna dict metriche (fitness = pnl + 50*win).""" series = spec.get("series", "A") p = spec["params"] o = data[series] if sl is not None: s, e = sl o = {k: v[s:e] for k, v in o.items()} rets = _backtest_grid(o, p, fee) nbars = len(o["close"]) months = nbars / data.get("bpm", TF_BPM["1h"]) if rets is None: # il bot si rifiuta di partire (vincolo break-even §4) return dict(n_trades=0, win_rate=0.0, pnl_pct=0.0, tpm=0.0, sharpe=0.0, avg_ret=0.0, qualified=False, refused=True, fitness=-2e6) n_tr = len(rets) tpm = n_tr / months if months > 0 else 0.0 if n_tr == 0: return dict(n_trades=0, win_rate=0.0, pnl_pct=0.0, tpm=0.0, sharpe=0.0, avg_ret=0.0, qualified=False, refused=False, fitness=-1e6) win_rate = float(np.mean(rets > 0)) pnl = float(np.sum(rets)) * 100 avg = float(np.mean(rets)) * 100 sharpe = float(np.mean(rets) / (np.std(rets) + 1e-12) * np.sqrt(tpm * 12)) \ if np.std(rets) > 0 else 0.0 qualified = tpm >= MIN_TRADES_PER_MONTH fitness = pnl + 50.0 * win_rate if not qualified: fitness = -1e6 + pnl return dict(n_trades=n_tr, win_rate=win_rate, pnl_pct=pnl, tpm=tpm, sharpe=sharpe, avg_ret=avg, qualified=qualified, refused=False, fitness=fitness) if __name__ == "__main__": import time data = load_anon("1h") print("loaded", data["n"], "bars,", data["dt"][0], "->", data["dt"][-1]) tr, va, te = splits3(data) demo = {"series": "B", "tf": "1h", "params": {"range_down": 0.10, "range_up": 0.10, "levels": 12, "sl_buf": 0.05, "tp_buf": 0.05, "max_bars": 1000}} t0 = time.time() print("TRAIN", evaluate(data, demo, tr)) print("VALID", evaluate(data, demo, va)) print("TEST ", evaluate(data, demo, te)) print("FULL ", evaluate(data, demo, None)) print(f"4 eval in {time.time()-t0:.2f}s")