diff --git a/.gitignore b/.gitignore index 3509a89..b9f78d0 100644 --- a/.gitignore +++ b/.gitignore @@ -18,6 +18,7 @@ data/processed/ notebooks/.ipynb_checkpoints/ data/paper_trades/ data/portfolio_paper/ +data/portfolio_paper_stats/ data/portfolios/ # stato locale di tooling (non condiviso) diff --git a/STRATEGIA_GRIGLIA.md b/STRATEGIA_GRIGLIA.md new file mode 100644 index 0000000..1a297e1 --- /dev/null +++ b/STRATEGIA_GRIGLIA.md @@ -0,0 +1,212 @@ +# Strategia di Grid Trading — Versione Corretta + +> Documento di specifica della strategia. Descrive *cosa* deve fare il bot e *perché*, +> non l'implementazione. È il riferimento da cui partire per riscrivere il codice in +> modo sicuro e testabile. + +--- + +## 1. Obiettivo + +Estrarre profitto dalla **volatilità di un asset all'interno di un intervallo di prezzo +(range)**, comprando automaticamente quando il prezzo scende e vendendo quando risale, +secondo livelli predefiniti (la "griglia"). + +La griglia **non prevede** la direzione del mercato: monetizza le oscillazioni. Funziona +quando il prezzo oscilla; perde quando il prezzo prende un trend deciso. Tutta la +progettazione che segue serve a massimizzare il primo caso e a limitare i danni nel +secondo. + +--- + +## 2. Principi corretti (cosa cambia rispetto al bot originale) + +| Aspetto | Bot originale (sbagliato) | Versione corretta | +|---|---|---| +| Asset | Shitcoin illiquida (LAND) | Coppia liquida (es. ETH/USDT, BNB/USDT) | +| Passo griglia | Assoluto in USDT (`GRID_STEP=3`) | **Percentuale** sul prezzo | +| Livelli | Mobili, inseguono il prezzo | **Fissi** dentro un range definito | +| Protezione perdite | Nessuna | **Stop-loss** sotto il range + **take-profit** sopra | +| Slippage | `amountOutMin = 0` (nessuna) | Calcolato da `getAmountsOut` − tolleranza | +| Break-even fee | Ignorato | Passo griglia **> costo round-trip** | +| Capitale | Tutto, senza limiti | Allocazione fissa, suddivisa per livello | +| Chiave privata | In chiaro nel `.env` | Keystore cifrato o input a runtime | +| Validazione | Nessuna | **Backtest** + **testnet** prima del capitale reale | + +--- + +## 3. Definizione della griglia + +### 3.1 Parametri di ingresso + +| Parametro | Significato | Esempio | +|---|---|---| +| `PAIR` | Coppia da tradare (base/quote) | `ETH/USDT` | +| `RANGE_LOW` | Estremo inferiore del range | `2800` | +| `RANGE_HIGH` | Estremo superiore del range | `3400` | +| `GRID_LEVELS` | Numero di livelli nella griglia | `12` | +| `CAPITAL_QUOTE` | Capitale totale in quote (USDT) | `1200` | +| `STOP_LOSS` | Prezzo sotto cui il bot chiude tutto e si ferma | `2650` | +| `TAKE_PROFIT` | Prezzo sopra cui il bot chiude tutto e si ferma | `3550` | +| `SLIPPAGE_BPS` | Tolleranza slippage in basis point (50 = 0,5%) | `50` | +| `FEE_BPS` | Fee del DEX in basis point (PancakeSwap = 25) | `25` | + +### 3.2 Costruzione dei livelli (griglia geometrica) + +I livelli vanno spaziati in **percentuale**, non in valore assoluto. Una griglia +geometrica mantiene lo stesso rendimento percentuale per ogni gradino, indipendentemente +dal prezzo. + +``` +ratio = (RANGE_HIGH / RANGE_LOW) ^ (1 / GRID_LEVELS) +livello[i] = RANGE_LOW * ratio ^ i per i = 0 .. GRID_LEVELS +``` + +Esempio (`RANGE_LOW=2800`, `RANGE_HIGH=3400`, `GRID_LEVELS=12`): +`ratio ≈ 1,0163` → passo di circa **1,63% per gradino**. + +### 3.3 Capitale per livello + +``` +quote_per_livello = CAPITAL_QUOTE / GRID_LEVELS +``` + +Ogni livello di acquisto impegna `quote_per_livello`. Il capitale è suddiviso in anticipo: +il bot **non** può comprare più di quanto allocato, e non scende mai sotto zero. + +--- + +## 4. Vincolo di break-even (regola anti-fee) + +Una griglia con passi troppo fitti perde: le fee di ogni round-trip (compra + vendi) si +mangiano il profitto. **Il passo percentuale della griglia deve superare il costo totale +di un round-trip.** + +``` +costo_round_trip ≈ 2 * (FEE_BPS + SLIPPAGE_BPS) / 10000 (in frazione) +passo_griglia = ratio - 1 + +VINCOLO: passo_griglia > costo_round_trip * margine_sicurezza (margine ≥ 1,5) +``` + +Esempio con `FEE_BPS=25`, `SLIPPAGE_BPS=50`: +`costo_round_trip ≈ 2 * (25+50)/10000 = 1,5%`. Con margine 1,5 → il passo deve essere +**≥ 2,25%**. Se la griglia geometrica dà 1,63%, **è troppo fitta**: vanno ridotti i +`GRID_LEVELS` o allargato il range finché il vincolo è rispettato. Il bot deve rifiutarsi +di partire se il vincolo non è soddisfatto. + +--- + +## 5. Logica operativa + +### 5.1 Inizializzazione +1. Validare i parametri: `RANGE_LOW < prezzo_attuale < RANGE_HIGH`, vincolo di break-even + rispettato, capitale disponibile sul wallet. +2. Verificare la coppia: liquidità sufficiente, contratto non honeypot (controllo + obbligatorio, non opzionale), token con decimali noti. +3. Costruire i livelli (§3.2) e marcare ognuno come `attivo`/`riempito`. +4. Allocare il capitale (§3.3). + +### 5.2 Ciclo principale +A ogni tick (es. ogni nuovo blocco, o ogni N secondi): + +``` +prezzo = prezzo_corrente(PAIR) + +# --- guardie di uscita: hanno priorità su tutto --- +SE prezzo <= STOP_LOSS: + vendi_tutta_la_posizione() # con slippage protetto + ferma il bot, log "STOP-LOSS" +SE prezzo >= TAKE_PROFIT: + vendi_tutta_la_posizione() + ferma il bot, log "TAKE-PROFIT" + +# --- logica griglia --- +PER ogni livello L: + SE prezzo attraversa L verso il basso E L non è ancora riempito: + compra quote_per_livello # con amountOutMin protetto + marca L come riempito + SE prezzo attraversa L verso l'alto E il livello sotto è riempito: + vendi la quantità di quel livello # con amountOutMin protetto + marca quel livello come libero +``` + +I livelli sono **fissi** (calcolati una volta), non inseguono il prezzo. Questo rende il +comportamento prevedibile e backtestabile. + +### 5.3 Calcolo `amountOutMin` (protezione slippage — obbligatoria) + +Mai passare `0`. Prima di ogni swap: + +``` +atteso = router.getAmountsOut(amountIn, path)[ultimo] +amountOutMin = atteso * (10000 - SLIPPAGE_BPS) / 10000 +``` + +Se la transazione non rientra nella tolleranza, deve **fallire** (revert), non eseguire a +qualsiasi prezzo. + +--- + +## 6. Gestione del rischio + +1. **Stop-loss obbligatorio** sotto `RANGE_LOW`. È la differenza tra "strategia" e + "gambling": senza stop-loss un trend ribassista svuota il wallet. +2. **Take-profit** sopra `RANGE_HIGH` per chiudere quando il prezzo esce dal range al + rialzo (la griglia avrebbe già venduto tutto; il take-profit evita di restare esposti). +3. **Capitale segregato**: usare un wallet dedicato, con solo il capitale destinato alla + strategia. Mai il wallet principale. +4. **Limite di gas e prezzo gas** ragionevoli, ricalcolati dinamicamente (no valori fissi + obsoleti). +5. **Kill-switch manuale**: comando per fermare il bot e liquidare in qualsiasi momento. +6. **Idempotenza/recupero**: se il bot si riavvia, deve ricostruire lo stato dei livelli + (riempiti/liberi) dal saldo on-chain, non ripartire da zero. + +--- + +## 7. Validazione prima del capitale reale + +Nessun fondo reale prima di aver superato, in ordine: + +1. **Backtest** su dati storici della coppia (almeno alcuni mesi, includendo sia fasi + laterali sia un trend marcato), misurando: PnL netto **dopo** fee e slippage, + max drawdown, numero di trade, comportamento allo stop-loss. +2. **Paper trading / simulazione** in tempo reale, senza eseguire ordini veri. +3. **Testnet** (BSC testnet) con la stessa logica e router di test, per verificare + l'esecuzione on-chain end-to-end. +4. **Mainnet con capitale minimo** (es. l'equivalente di pochi euro) per la prima + settimana, poi scalare solo se i risultati combaciano col backtest. + +--- + +## 8. Quando NON usare questa strategia + +- Asset illiquido o a rischio rug-pull/honeypot. +- Mercato in trend forte e prolungato (la griglia perde: lo stop-loss limita il danno ma + non genera profitto). +- Passo griglia che non rispetta il vincolo di break-even (§4). +- Capitale che non puoi permetterti di perdere. + +--- + +## 9. Parametri di esempio (configurazione di partenza prudente) + +``` +PAIR = BNB/USDT # coppia liquida su PancakeSwap +RANGE_LOW = 580 +RANGE_HIGH = 720 +GRID_LEVELS = 8 # passo ≈ 2,7% > break-even +CAPITAL_QUOTE = 400 # USDT, su wallet dedicato +STOP_LOSS = 545 # ~6% sotto RANGE_LOW +TAKE_PROFIT = 760 # ~5,5% sopra RANGE_HIGH +SLIPPAGE_BPS = 50 # 0,5% +FEE_BPS = 25 # PancakeSwap v2 +``` + +Verifica break-even: passo ≈ 2,7% > `1,5 × (2×0,75%) = 2,25%` ✅ + +--- + +*Questo documento descrive la strategia. L'implementazione (ethers v6, gestione sicura +della chiave, calcolo slippage, stato persistente, backtester) va sviluppata a parte, +con test, e validata secondo §7 prima di qualsiasi capitale reale.* diff --git a/docs/diary/2026-06-10-grid-traders-game3.md b/docs/diary/2026-06-10-grid-traders-game3.md new file mode 100644 index 0000000..04cd032 --- /dev/null +++ b/docs/diary/2026-06-10-grid-traders-game3.md @@ -0,0 +1,113 @@ +# 2026-06-10 — Gioco "Blind Traders" sessione 3: GRID TRADERS (regola: STRATEGIA_GRIGLIA.md) + +## Setup + +Terza sessione del gioco dei trader ciechi, con una regola nuova: ogni agente deve +implementare la strategia descritta in `STRATEGIA_GRIGLIA.md` (grid trading a griglia +geometrica). 100 agenti (haiku, via Workflow) ricevono SOLO un digest anonimo di due +serie X/Y (in realtà BTC/ETH, mai rivelato) sul loro timing assegnato +(25×15m, 20×30m, 20×1h, 15×2h, 12×4h, 8×1d) + 5 archetipi di stile a rotazione +(prudente / aggressivo / asimmetrico-rialzista / asimmetrico-ribassista / data-driven), +e propongono la CONFIGURAZIONE della griglia: `range_down/up %`, `grid_levels`, +`sl_buf`, `tp_buf`, `max_bars`, serie. + +Infrastruttura nuova in `scripts/games/`: +- **`grid_engine.py`** — backtest deterministico, causale, fee-aware della spec: + griglia geometrica `ratio=((1+ru)/(1-rd))^(1/L)` costruita sul close di deploy, + capitale 1/L per livello (§3.3), buy su attraversamento ↓ di un livello non + riempito, sell del livello su attraversamento ↑ del successivo (§5.2), SL sotto + il range e TP sopra che liquidano tutto (§6), redeploy a fine episodio + (SL/TP/max_bars). **Vincolo break-even §4 implementato alla lettera**: passo + ≤ 1.5×costo RT → il motore si rifiuta di partire (`refused`, fitness −2e6); + `_normalize` dell'arena riduce i livelli al massimo legale. Fill intrabar lungo + il percorso O→L→H→C / O→H→L→C; fee 0.10% RT per round-trip + slippage opzionale. +- **`grid_arena.py`** — torneo identico alle sessioni 1-2: split 60/20/20 + TRAIN/VALID/TEST, 90 epoche di hill-climb sul TRAIN, cull del 10% in VALID ogni + 10 epoche → 10 superstiti; TEST = OOS puro mai ottimizzato. +- **`grid_brief.py`** — digest anonimo con statistiche per dimensionare una griglia: + escursione max/min rolling (w100/500/2000, mediana e p90) e probabilità di fuga + da un range ±5/10/20% entro 500 barre. + +## Esito + +`data/games/grid_result.json` (+ log `grid_tournament.log`). 100/100 spec da agenti +reali (nessun sostituto random). + +- **Alla proposta, 80/100 agenti scelgono X (=BTC)**: dai soli numeri anonimi capiscono + che la griglia sopravvive meglio sulla serie meno volatile (escape ±20% in 500 barre: + BTC 34.5% vs ETH 53.2% a 1h). +- **L'evoluzione ribalta la scelta: tutti i 10 superstiti finiscono su Y (=ETH)** — + nel periodo VALID/TEST la vol più alta di ETH paga di più i round-trip, e il rischio + trend è gestito non dal range stretto ma dalla FORMA della griglia (sotto). +- **Convergenza fortissima della forma** (9/10 superstiti): griglia **asimmetrica + ribassista** — range profondo sotto (−13/−20%), corto sopra (+4/+8%), **livelli al + minimo (4)** → passo largo ~4.5-5.5%, SL buffer profondo (5-15%), max_bars lunghi. + Tradotto: **compra i dip di ETH in 4 tranche distanziate ~5%, rivendi ogni tranche + al rimbalzo di un passo, stop catastrofale a ~−25/−30%**. Il gioco ha ri-scoperto + per la terza volta la mean-reversion (qui in forma di dip-buying a tranche), e ha + imparato da solo la lezione anti-fee: meglio pochi passi larghi che griglie fitte. + +**Vincitore: agente #11, ETH 15m**, griglia −17.1%/+4.6% × 4 livelli, SL buf 12.4%, +TP buf 4.8%, max 2143 barre. TEST(OOS): **PnL +891% (additivo), win 97%, 38.5 +trade/mese, Sharpe 10.1**. FULL 2018-2026 (include bear 2018 e 2022): +4284%, Sharpe 9.6. +Stress slippage TEST: 0.05%/lato → 871%, 0.10%/lato → 850% (il passo ~5.4% ⋙ costi). + +## Caveat onesti (perché NON è un candidato deploy così com'è) + +1. **Il PnL è additivo per-trade e non misura il drawdown UNREALIZED**: l'engine + somma i round-trip realizzati; mentre la griglia tiene 4 tranche dentro un + drawdown −17%, l'equity vera è sott'acqua (fino a ~−15% di episodio + SL −25/30% + quando scatta). Il win-rate 97% è il profilo classico della griglia: tante + micro-vincite, perdite rare ma grandi (stessa famiglia di rischio del + short-vol/martingala). Per un gate serio servirebbe l'equity curve mark-to-market. +2. **Long-only su ETH con VALID/TEST 2023-2026** (regime rialzista/oscillante): + l'asimmetria −17/+4.6 è anche un fit al regime. Il FULL positivo (bear inclusi) + è incoraggiante ma il grosso del PnL resta nei periodi di reversione rialzista. +3. Le candele flat ETH 15m (14-30%/anno) e i fill intrabar "al livello" condividono + i caveat noti del progetto (bias pro-stop-stretti dell'engine intrabar; qui gli + stop sono larghi, quindi l'effetto è minore). + +## Lezioni + +- La spec STRATEGIA_GRIGLIA.md è implementabile in modo causale e onesto; il suo + vincolo break-even (§4) è esattamente la "lezione fee" del progetto, e il torneo + l'ha confermato spingendo i livelli al minimo (passo massimo). +- La griglia evoluta è un parente povero delle fade MR già in live: stesso edge + (reversione ETH), ma incassato con inventory risk crescente invece che con + TP/SL per-trade. Non sostituisce le MR; eventuale interesse solo come variante + "a tranche" da gateare con equity mark-to-market (e confronto con MR01/MR02 a + parità di rischio) PRIMA di pensarci per il PORT06. + +## GATE "si puo' inserire?" (stessa sera) — NO-GO: edge = artefatto delle wick testnet + +`scripts/analysis/grid_game_gate.py`: engine **mark-to-market** dedicato (equity +per barra = capitale + inventario al close, SL gap-aware, flat-skip, fee 0.10% RT), +metriche standard del progetto + gate PORT06. Primo passaggio ingannevole: standalone +WINNER 15m FULL Sharpe 5.61 DD 15.8%, corr max coi sleeve 0.34, plateau 16/16, +e il criterio formale promuoveva il half-size (OOS Sh 10.07->10.12, DD 1.48->1.36). +MA il breakdown annuale mostrava **+4946% nel 2022** (ETH −70%): impossibile. + +**Verifica avversariale decisiva**: il feed Deribit (testnet) e' pieno di **spike +print** — ETH 15m ha 1064 barre con wick ≥6% sotto i close adiacenti che rientrano +subito (fino a −19% in 15 minuti; BTC ha spike a −54% nel feb 2024). La griglia +intrabar compra su quelle wick FINTE e rivende al rimbalzo: free money in sim, +infillabile dal vivo. Stress **close-only** (fill solo su attraversamento del +close): WINNER 15m CAGR 1544%->21%, Sharpe 5.61->0.92, OOS +2738%->+32%, +trades 3803->1156; top3 1h CAGR -1.7%, OOS −18%. **~99% dell'edge vive nelle +wick.** La versione onesta (Sharpe 0.92, DD 27%) e' molto sotto ogni sleeve +deployato → **NON inseribile**, in nessuna size. + +Lezioni aggiuntive: +- Il torneo ha massimizzato esattamente l'artefatto: la convergenza su ETH 15m + range profondo era la firma del **wick harvesting**, non di un edge. +- I motori intrabar dei giochi (`grid_engine`, e in misura minore i TP intrabar + di `engine.py`) vanno SEMPRE stressati con una variante close-only prima di + promuovere qualunque vincitore: aggiungere il check ai prossimi giochi. +- Gli sleeve in produzione NON sono toccati da questo artefatto nello stesso modo: + pairs entra/esce sui close, le fade hanno EXIT-16 close-confirm sullo SL, e + soprattutto il ledger REAL-TRUTH usa i fill reali (che gli spike non fillano). + +Artefatti: `scripts/games/grid_{engine,arena,brief}.py`, spec agenti in +`data/games/specs_grid/`, digest `data/games/grid_digests.json`, risultato +`data/games/grid_result.json`, log `data/games/grid_tournament.log`, gate +`scripts/analysis/grid_game_gate.py` (param `close_only` per lo stress anti-wick). diff --git a/scripts/analysis/grid_game_gate.py b/scripts/analysis/grid_game_gate.py new file mode 100644 index 0000000..9e7c694 --- /dev/null +++ b/scripts/analysis/grid_game_gate.py @@ -0,0 +1,322 @@ +"""GATE PORT06 — griglia ETH (vincitore gioco "Grid Traders", sessione 3). + +Il gioco (scripts/games/grid_*, regola STRATEGIA_GRIGLIA.md) ha promosso una +griglia geometrica asimmetrica su ETH: range profondo sotto, corto sopra, +4 livelli (passo ~5%), SL catastrofale. Ma il motore del gioco somma i PnL +REALIZZATI per trade e NON misura l'equity mark-to-market: l'inventario a +tranche dentro un drawdown e' rischio vero che il fitness non vede. + +Questo gate risponde alla domanda "si puo' inserire?" con il metodo del progetto: + [1] STANDALONE mark-to-market (engine MTM dedicato, fill onesti): + equity per barra = capitale + inventario valutato al close; fee 0.10% RT + (taker; i fill ai livelli sarebbero LIMIT->maker, quindi conservativo); + SL gap-aware (gap sotto lo stop -> fill all'open, non al livello); + flat-skip (nessun fill sulle candele O=H=L=C di ETH 15m, live-realizable). + Metriche FULL/OOS con le stesse funzioni degli altri gate + stress fee 2x. + [2] CORRELAZIONE coi 19 sleeve PORT06 (il sospetto: e' la stessa reversione + ETH delle fade MR, incassata con inventory risk). + [3] ROBUSTEZZA: plateau range_down x range_up attorno al vincitore. + [4] GATE PORT06: baseline vs +GRID (full e half size). Promosso solo se + OOS Sharpe non peggiora E DD non sale (criterio standard). + + uv run python scripts/analysis/grid_game_gate.py +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.combine_portfolio import port_returns, metrics, SPLIT, OOS_DATE, IDX +from scripts.portfolios._defs import PORTFOLIOS +from src.portfolio import weighting as W +from src.data.downloader import load_data + +POS, LEV = 0.15, 3.0 # config canonica sleeve (== build_everything) +FEE_SIDE = 0.0005 # 0.05%/lato = 0.10% RT + +# top-3 del torneo (data/games/grid_result.json) +WINNER_15M = dict(tf="15m", range_down=0.171, range_up=0.046, levels=4, + sl_buf=0.124, tp_buf=0.048, max_bars=2143) +TOP2_30M = dict(tf="30m", range_down=0.158, range_up=0.048, levels=4, + sl_buf=0.081, tp_buf=0.044, max_bars=613) +TOP3_1H = dict(tf="1h", range_down=0.134, range_up=0.053, levels=4, + sl_buf=0.150, tp_buf=0.063, max_bars=562) + +_RESAMPLE = {"30m": ("15m", "30min")} + + +def _load(asset, tf): + if tf in _RESAMPLE: + base, rule = _RESAMPLE[tf] + d = load_data(asset, base).copy() + d["dt"] = pd.to_datetime(d["datetime"]) + g = d.set_index("dt").resample(rule).agg( + {"open": "first", "high": "max", "low": "min", "close": "last", + "volume": "sum"}).dropna(subset=["open", "close"]).reset_index() + g["datetime"] = g["dt"] + return g + d = load_data(asset, tf).copy() + d["dt"] = pd.to_datetime(d["datetime"]) + return d + + +def grid_mtm(asset="ETH", *, tf, range_down, range_up, levels, sl_buf, tp_buf, + max_bars, pos=POS, lev=LEV, fee_side=FEE_SIDE, flat_skip=True, + close_only=False): + """Griglia STRATEGIA_GRIGLIA.md con contabilita' mark-to-market. + + Ritorna (equity daily Series base 1.0, stats dict). Causale: deploy sul + close, fill dalle barre successive lungo il percorso O->L->H->C / O->H->L->C. + """ + df = _load(asset, tf) + op = df["open"].to_numpy(float) + hi = df["high"].to_numpy(float) + lo = df["low"].to_numpy(float) + cl = df["close"].to_numpy(float) + dt = pd.to_datetime(df["datetime"]).to_numpy() + n = len(cl) + ratio = ((1 + range_up) / (1 - range_down)) ** (1.0 / levels) + if ratio - 1 <= 1.5 * 2 * fee_side: # vincolo break-even §4 + raise ValueError("break-even violato") + flat = (op == hi) & (op == lo) & (op == cl) + + capital = 1.0 + eq = np.empty(n) + eq[:20] = 1.0 + # stato episodio + active = False + lv = []; filled = []; tn = 0.0; sl = tp = 0.0; ep_end = 0 + n_open = 0 + trades = wins = stops = 0 + deploy_i = -1 + + def mtm(px): + u = 0.0 + for k in range(levels): + if filled[k]: + u += tn * (px / lv[k] - 1.0) + return capital + u + + i = 20 + for j in range(20, n): + if not active: + # deploy sul close di j (fill da j+1) + px = cl[j] + rl_ = px * (1 - range_down) + lv = [rl_ * ratio ** k for k in range(levels + 1)] + sl = rl_ * (1 - sl_buf) + tp = lv[levels] * (1 + tp_buf) + filled = [False] * levels + n_open = 0 + tn = capital * pos * lev / levels # notional per tranche + ep_end = j + max_bars + active = True + deploy_i = j + eq[j] = capital + continue + if flat_skip and flat[j]: + eq[j] = mtm(cl[j]) + continue + cur = cl[j - 1] + if close_only: + # fill solo su attraversamento del CLOSE (le wick non fillano): + # stress anti-spike-print del feed testnet + pts = (cl[j],) + else: + pts = (op[j], lo[j], hi[j], cl[j]) if cl[j] >= op[j] \ + else (op[j], hi[j], lo[j], cl[j]) + died = False + for pi, q in enumerate(pts): + q = float(q) + if q == cur: + continue + if q < cur: + from bisect import bisect_left + k1 = bisect_left(lv, q) + k2 = bisect_left(lv, cur) - 1 + for k in range(min(k2, levels - 1), max(k1, 0) - 1, -1): + if not filled[k]: + filled[k] = True + n_open += 1 + capital -= fee_side * tn # fee ingresso + if q <= sl: + # STOP: gap all'open -> fill all'open, altrimenti al livello sl + fill = q if (pi == 0 and q <= sl) else sl + for k in range(levels): + if filled[k]: + r = fill / lv[k] + capital += tn * (r - 1.0) - fee_side * tn * r + filled[k] = False + trades += 1 + stops += 1 + n_open = 0 + died = True + cur = q + break + else: + from bisect import bisect_right + m1 = bisect_right(lv, cur) + m2 = bisect_right(lv, q) - 1 + for m in range(max(m1, 1), min(m2, levels) + 1): + k = m - 1 + if filled[k]: + r = lv[m] / lv[k] + capital += tn * (r - 1.0) - fee_side * tn * r + filled[k] = False + n_open -= 1 + trades += 1 + wins += 1 + if q >= tp: + for k in range(levels): + if filled[k]: + r = tp / lv[k] + capital += tn * (r - 1.0) - fee_side * tn * r + filled[k] = False + trades += 1 + wins += 1 + n_open = 0 + died = True + cur = q + break + cur = q + if not died and j >= ep_end: + # timeout: liquida al close + for k in range(levels): + if filled[k]: + r = cl[j] / lv[k] + capital += tn * (r - 1.0) - fee_side * tn * r + filled[k] = False + trades += 1 + wins += r > 1.0 + 2 * fee_side + n_open = 0 + died = True + if died: + active = False + eq[j] = capital + else: + eq[j] = mtm(cl[j]) + + s = pd.Series(eq, index=pd.DatetimeIndex(dt)).resample("1D").last().dropna() + s = s / s.iloc[0] + return s, dict(trades=trades, win=100.0 * wins / max(1, trades), stops=stops) + + +def std(eqd): + """Metriche FULL/OOS con le funzioni standard del progetto.""" + e = eqd.reindex(IDX).ffill().bfill() + dr = e.pct_change().fillna(0.0) + return metrics(dr), metrics(dr, lo=SPLIT) + + +def main(): + p = PORTFOLIOS["PORT06"] + print("=" * 100) + print(" GATE PORT06 — griglia ETH (vincitore gioco Grid Traders) | " + f"pos={POS} lev={LEV} | OOS da {OOS_DATE}") + print("=" * 100) + + # [1] STANDALONE mark-to-market + print("\n[1] STANDALONE mark-to-market (fee 0.10% RT, flat-skip, SL gap-aware):") + print(f" {'cfg':<22s}{'trd':>7s}{'win%':>6s}{'stops':>6s}{'FULL%':>8s}{'CAGR%':>7s}" + f"{'DD%':>7s}{'Shrp':>6s} | {'OOS%':>7s}{'oDD%':>6s}{'oShrp':>6s}") + eqs = {} + for tag, cfg in [("WINNER 15m", WINNER_15M), ("top2 30m", TOP2_30M), + ("top3 1h", TOP3_1H)]: + eqd, st = grid_mtm("ETH", **cfg) + f, o = std(eqd) + eqs[tag] = eqd + print(f" {tag:<22s}{st['trades']:>7d}{st['win']:>6.1f}{st['stops']:>6d}" + f"{f['ret']:>+8.0f}{f['cagr']:>7.1f}{f['dd']:>7.2f}{f['sharpe']:>6.2f}" + f" | {o['ret']:>+7.0f}{o['dd']:>6.2f}{o['sharpe']:>6.2f}") + # stress: fee 2x e no flat-skip sul winner + eq2, st2 = grid_mtm("ETH", **WINNER_15M, fee_side=0.001) + f2, o2 = std(eq2) + eqnf, _ = grid_mtm("ETH", **WINNER_15M, flat_skip=False) + fnf, onf = std(eqnf) + print(f" {'winner fee 2x':<22s}{st2['trades']:>7d}{st2['win']:>6.1f}{'':>6s}" + f"{f2['ret']:>+8.0f}{f2['cagr']:>7.1f}{f2['dd']:>7.2f}{f2['sharpe']:>6.2f}" + f" | {o2['ret']:>+7.0f}{o2['dd']:>6.2f}{o2['sharpe']:>6.2f}") + print(f" {'winner no-flat-skip':<22s}{'':>7s}{'':>6s}{'':>6s}" + f"{fnf['ret']:>+8.0f}{fnf['cagr']:>7.1f}{fnf['dd']:>7.2f}{fnf['sharpe']:>6.2f}" + f" | {onf['ret']:>+7.0f}{onf['dd']:>6.2f}{onf['sharpe']:>6.2f}") + + grid_eq = eqs["WINNER 15m"] + + # [2] CORRELAZIONI coi sleeve PORT06 + from src.portfolio.sleeves import all_sleeve_equities + eq_base = dict(all_sleeve_equities()) + gr = grid_eq.reindex(IDX).ffill().bfill().pct_change().fillna(0.0) + print("\n[2] CORRELAZIONE rendimenti giornalieri GRID_ETH15M vs sleeve PORT06 (top 8):") + cors = {} + for sid, e in eq_base.items(): + r = e.reindex(IDX).ffill().bfill().pct_change().fillna(0.0) + cors[sid] = gr.corr(r) + for sid, cv in sorted(cors.items(), key=lambda kv: -abs(kv[1]))[:8]: + print(f" {sid:<16s} {cv:+.3f}") + + # [3] ROBUSTEZZA: plateau range_down x range_up (15m, levels=4) + print("\n[3] ROBUSTEZZA 15m (Sharpe FULL mark-to-market, levels=4, " + "sl_buf=0.12, tp_buf=0.05, max_bars=2000):") + rds = [0.13, 0.15, 0.17, 0.19] + rus = [0.04, 0.05, 0.06, 0.08] + print(" rd\\ru " + "".join(f"{ru:>7.2f}" for ru in rus)) + cells = tot = 0 + for rd in rds: + row = f" {rd:>5.2f} " + for ru in rus: + eqd, _ = grid_mtm("ETH", tf="15m", range_down=rd, range_up=ru, + levels=4, sl_buf=0.12, tp_buf=0.05, max_bars=2000) + f, o = std(eqd) + tot += 1 + cells += (f["sharpe"] > 1) and (o["sharpe"] > 1) + row += f"{f['sharpe']:>7.2f}" + print(row) + print(f" -> {cells}/{tot} celle con Sharpe>1 sia FULL che OOS") + + # [4] GATE PORT06 + print("\n[4] GATE PORT06 — baseline vs +GRID_ETH15M (full/half size):") + def port_m(extra=None): + members = dict(eq_base) + ids = list(p.sleeve_ids) + if extra is not None: + members["GRID_ETH15M"] = extra + ids = ids + ["GRID_ETH15M"] + dr = pd.DataFrame({i: members[i].reindex(IDX).ffill().bfill() + .pct_change().fillna(0.0) for i in ids}) + w = W.weight_vector(p.weighting, ids, dr, weights=p.weights, + caps=p.caps, clusters=p.clusters, lookback=p.vol_lookback) + drp = port_returns({i: members[i].reindex(IDX).ffill().bfill() + for i in ids}, w) + return metrics(drp), metrics(drp, lo=SPLIT) + + half = (1 + 0.5 * gr).cumprod() + res = {"baseline": port_m(None), + "+GRID full": port_m(grid_eq), + "+GRID half": port_m(half)} + print(f" {'variante':<12s} | {'FULL Sh':>8s}{'FULL DD%':>9s}{'CAGR%':>7s}" + f" | {'OOS Sh':>7s}{'OOS DD%':>8s}") + for tag, (f, o) in res.items(): + print(f" {tag:<12s} | {f['sharpe']:>8.2f}{f['dd']:>9.2f}{f['cagr']:>7.1f}" + f" | {o['sharpe']:>7.2f}{o['dd']:>8.2f}") + + fb, ob = res["baseline"] + print("\n" + "=" * 100) + print(" VERDETTO (criterio standard: OOS Sharpe non peggiora E DD non sale)") + print("=" * 100) + for tag in ("+GRID full", "+GRID half"): + f, o = res[tag] + ok = (o["sharpe"] >= ob["sharpe"] - 0.02 and o["dd"] <= ob["dd"] + 1e-9 + and f["sharpe"] >= fb["sharpe"] - 0.02) + print(f" {tag:<12s}: OOS Sh {ob['sharpe']:.2f}->{o['sharpe']:.2f} " + f"DD {ob['dd']:.2f}->{o['dd']:.2f} | FULL Sh {fb['sharpe']:.2f}->{f['sharpe']:.2f} " + f"DD {fb['dd']:.2f}->{f['dd']:.2f} => {'PROMOSSO' if ok else 'bocciato'}") + + +if __name__ == "__main__": + main() diff --git a/scripts/analysis/pairs30m_gate.py b/scripts/analysis/pairs30m_gate.py new file mode 100644 index 0000000..74bb960 --- /dev/null +++ b/scripts/analysis/pairs30m_gate.py @@ -0,0 +1,92 @@ +"""GATE: aggiungere ETH/BTC 30m (vincitore gioco sessione 2) AL BLEND attuale (1h+15m)? + +Domanda: il 30m e' un 3o timeframe utile dello spread ETH/BTC, o e' ridondante col 15m +adiacente gia' deployato? Test (engine pairs_sim_flat, == worker): + [1] correlazioni: 30m vs 1h, 30m vs 15m (se ~1 col 15m -> ridondante). + [2] gate PORT06: baseline ATTUALE (6 pairs, incl 15m) vs +30m (7 pairs), mezza size. + [3] robustezza ai costi (fee sweep) del 30m. + + uv run python scripts/analysis/pairs30m_gate.py +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.combine_portfolio import port_returns, metrics, SPLIT, OOS_DATE +from scripts.analysis.pairs_research import pairs_sim, pairs_sim_flat +from scripts.analysis.report_families import daily_from +from scripts.portfolios._defs import PORTFOLIOS +from src.portfolio.sleeves import all_sleeve_equities +from src.portfolio import weighting as W + +WIN_30M = dict(n=53, z_in=1.947, z_exit=1.0, max_bars=24) # vincitore gioco sess.2 + + +def daily(cfg, tf, flat_skip, pos=0.15): + if flat_skip: + r = pairs_sim_flat("ETH", "BTC", tf=tf, **cfg, flat_skip=True, pos=pos) + else: + r = pairs_sim("ETH", "BTC", tf=tf, **cfg, pos=pos) + return daily_from(r["eq_ts"], r["eq_v"]), r + + +def port_metrics(members, ids, clusters, caps): + dr = pd.DataFrame({i: members[i].pct_change().fillna(0.0) for i in ids}) + w = W.weight_vector("cap", ids, dr, caps=caps, clusters=clusters) + drp = port_returns({i: members[i] for i in ids}, w) + return metrics(drp), metrics(drp, lo=SPLIT), w + + +def main(): + p = PORTFOLIOS["PORT06"] + eq_base = dict(all_sleeve_equities()) # include gia' PR_ETHBTC (1h) e PR_ETHBTC_15M + e1h = eq_base["PR_ETHBTC"] + e15 = eq_base["PR_ETHBTC_15M"] + e30h, r30 = daily(WIN_30M, "30m", flat_skip=True, pos=0.075) # half size come il 15m + + print("=" * 92) + print(" GATE — ETH/BTC 30m (vincitore gioco sess.2) sopra il BLEND 1h+15m attuale") + print(f" 30m: {r30['trades']} trade, {r30.get('n_skip_entry',0)} ingressi flat saltati") + print("=" * 92) + + def corr(a, b): return a.pct_change().fillna(0).corr(b.pct_change().fillna(0)) + print("\n[1] CORRELAZIONI (rendimenti giornalieri):") + print(f" 30m vs 1h : {corr(e30h, e1h):.3f}") + print(f" 30m vs 15m: {corr(e30h, e15):.3f} <-- se alta, ridondante col 15m gia' deployato") + print(f" (rif) 15m vs 1h: {corr(e15, e1h):.3f}") + + print(f"\n[2] GATE PORT06 (cap PAIRS 0.33 + SHAPE 0.0588) | OOS da {OOS_DATE}:") + ids0 = list(p.sleeve_ids) + cl0 = p.clusters + caps = p.caps + f0, o0, _ = port_metrics(eq_base, ids0, cl0, caps) + # + 30m + mem1 = dict(eq_base); mem1["PR_ETHBTC_30M"] = e30h + ids1 = ids0 + ["PR_ETHBTC_30M"] + cl1 = dict(cl0); cl1["PR_ETHBTC_30M"] = "ETH-rev" + f1, o1, w1 = port_metrics(mem1, ids1, cl1, caps) + print(f" {'config':<22}{'FULL Sh':>8}{'FULL DD%':>9}{'OOS Sh':>8}{'OOS DD%':>8}") + print(f" {'ATTUALE (1h+15m)':<22}{f0['sharpe']:>8.2f}{f0['dd']:>9.2f}{o0['sharpe']:>8.2f}{o0['dd']:>8.2f}") + print(f" {'+30m (1h+15m+30m)':<22}{f1['sharpe']:>8.2f}{f1['dd']:>9.2f}{o1['sharpe']:>8.2f}{o1['dd']:>8.2f}") + ok = o1["sharpe"] >= o0["sharpe"] - 0.02 and o1["dd"] <= o0["dd"] + 1e-9 \ + and f1["sharpe"] >= f0["sharpe"] - 0.02 and f1["dd"] <= f0["dd"] + 1e-9 + print(f" => {'MIGLIORA (promosso)' if ok else 'NON migliora (bocciato)'}") + print(f" peso pairs ETH/BTC: 1h {w1.get('PR_ETHBTC',0)*100:.1f}% + 15m " + f"{w1.get('PR_ETHBTC_15M',0)*100:.1f}% + 30m {w1.get('PR_ETHBTC_30M',0)*100:.1f}% " + f"= {(w1.get('PR_ETHBTC',0)+w1.get('PR_ETHBTC_15M',0)+w1.get('PR_ETHBTC_30M',0))*100:.1f}% su 7 coppie") + + print("\n[3] ROBUSTEZZA AI COSTI (30m standalone, Sharpe per fee RT/coppia):") + for fee, lbl in [(0.001, "0.20% 1x"), (0.002, "0.40% 2x"), (0.003, "0.60% 3x"), + (0.004, "0.80% 4x"), (0.006, "1.20% 6x")]: + r = pairs_sim_flat("ETH", "BTC", tf="30m", **WIN_30M, flat_skip=True, fee_rt=fee) + print(f" {lbl:<10} Sharpe {r['sharpe']:>6.2f}") + + +if __name__ == "__main__": + main() diff --git a/scripts/analysis/reset_flatten.py b/scripts/analysis/reset_flatten.py new file mode 100644 index 0000000..80dc7b0 --- /dev/null +++ b/scripts/analysis/reset_flatten.py @@ -0,0 +1,64 @@ +"""Flatten one-shot del conto Deribit testnet per il RESET del portafoglio (2026-06-10). + +Cancella gli ordini resting noti (TP limit / disaster-SL dai status.json dei worker), +poi chiude TUTTE le posizioni USDC con market reduce-only (close_position di cerbero +non supporta i lineari USDC) e verifica che il conto sia flat. + +Uso: uv run python scripts/analysis/reset_flatten.py +""" +import glob +import json +import time + +from src.live.cerbero_client import CerberoClient +from src.live.execution import ExecutionClient, contract_spec, register_contract + + +def main(): + c = CerberoClient() + ex = ExecutionClient(client=c) + + # 1) ordini resting noti dai status.json (TP limit + disaster bracket) + oids = set() + for f in glob.glob("data/portfolio_paper/*/status.json"): + s = json.load(open(f)) + for k in ("real_tp_order_id", "real_dsl_order_id"): + if s.get(k): + oids.add(s[k]) + for oid in sorted(oids): + r = ex.cancel_order(oid) + print(f"cancel {oid}: {r.get('state', r)}") + + # 2) flatten posizioni USDC (market reduce-only, max 5 passate) + for attempt in range(5): + live = [p for p in c.get_positions("USDC") + if abs(float(p.get("size", 0) or 0)) > 0] + if not live: + print("FLAT — conto pulito") + return True + for p in live: + inst = p["instrument"] + register_contract(inst, c) + size_usd = abs(float(p["size"])) + mark = float(p.get("mark_price") or 0) + step = contract_spec(inst)["step"] + entry_side = "buy" if p["direction"] == "long" else "sell" + units = size_usd / mark if mark else 0 + amount = round(units / step) * step + if amount <= 0: + print(f" {inst}: residuo {size_usd:.2f} USD sotto mezzo step, ok") + continue + f = ex.close_amount(inst, entry_side, amount, label="reset_flatten") + print(f" close {inst} amt={amount} -> {f.order_state} fill={f.fill_price} " + f"verified={f.verified} {f.notes[:60]}") + time.sleep(3) + + left = [(p["instrument"], round(float(p["size"]), 4)) + for p in c.get_positions("USDC") if abs(float(p.get("size", 0) or 0)) > 0] + print("RESIDUO FINALE:", left or "FLAT") + return not left + + +if __name__ == "__main__": + ok = main() + raise SystemExit(0 if ok else 1) diff --git a/scripts/games/arena.py b/scripts/games/arena.py index 0b33023..da4c135 100644 --- a/scripts/games/arena.py +++ b/scripts/games/arena.py @@ -46,6 +46,19 @@ SINGLE_FAMILIES = ["zscore", "breakout", "ma_cross", "rsi", "momentum"] DIRECTIONS = ["reversion", "trend"] TIMEFRAMES = ["5m", "15m", "30m", "1h", "2h", "4h", "1d"] # tutti i timing validi +# Vincolo opzionale: accetta SOLO strategie NON gia' usate in live. Firme live (da +# vietare): 'pairs' (PR01) + REVERSION di zscore(MR01)/breakout(MR02)/momentum(MR07, +# return-reversal). NB: momentum+reversion == MR07 -> e' LIVE, va vietato (loophole). +# Coercizione: pairs -> ma_cross(trend); (zscore|breakout|momentum)+reversion -> +trend. +# Resta spazio NUOVO: trend di zscore/breakout/momentum, ma_cross, rsi (ogni direzione). +NO_LIVE = False +_LIVE_REV_FAMS = {"zscore", "breakout", "momentum"} # in reversion = MR01/MR02/MR07 live + + +def set_no_live(v: bool): + global NO_LIVE + NO_LIVE = bool(v) + def _rand_param(rng, lo, hi, typ): if typ == "i": @@ -54,7 +67,7 @@ def _rand_param(rng, lo, hi, typ): def random_spec(rng): - if rng.random() < 0.25: + if not NO_LIVE and rng.random() < 0.25: fam = "pairs" else: fam = rng.choice(SINGLE_FAMILIES) @@ -66,7 +79,10 @@ def random_spec(rng): spec["series"] = "AB" else: spec["series"] = rng.choice(["A", "B"]) - spec["params"]["direction"] = rng.choice(DIRECTIONS) + d = rng.choice(DIRECTIONS) + if NO_LIVE and fam in _LIVE_REV_FAMS: + d = "trend" # zscore/breakout in reversion sono live -> trend + spec["params"]["direction"] = d return spec @@ -98,6 +114,8 @@ def _normalize(spec): fam = spec.get("family") if fam not in SPACE: fam = "zscore" + if NO_LIVE and fam == "pairs": + fam = "ma_cross" # pairs (PR01) e' live -> rimpiazza con ma_cross (nuovo, trend) out = {"family": fam, "params": {}} for k, (lo, hi, typ) in SPACE[fam].items(): v = spec.get("params", {}).get(k, (lo + hi) / 2) @@ -113,7 +131,10 @@ def _normalize(spec): else: out["series"] = spec.get("series", "A") if spec.get("series") in ("A", "B") else "A" d = spec.get("params", {}).get("direction") or spec.get("direction") - out["params"]["direction"] = d if d in DIRECTIONS else "reversion" + d = d if d in DIRECTIONS else "reversion" + if NO_LIVE and fam in _LIVE_REV_FAMS and d == "reversion": + d = "trend" # zscore/breakout in reversion = fade live -> trend + out["params"]["direction"] = d return out @@ -165,12 +186,12 @@ def run_tournament(specs, briefs=None, seed=7, for ep in range(1, epochs + 1): # elaborazione: l'agente affina sul TRAIN (cio' che vede); ricalcola VALID for a in alive(): - cand = mutate(a.spec, rng) + cand = _normalize(mutate(a.spec, rng)) # normalizza PRIMA di valutare data = datasets[a.tf] tr, va, _ = splits_map[a.tf] m = evaluate(data, cand, tr) if m["fitness"] > a.train_fit: - a.spec = _normalize(cand) + a.spec = cand a.metrics, a.train_fit = m, m["fitness"] a.vmetrics = evaluate(data, a.spec, va) a.valid_fit = a.vmetrics["fitness"] diff --git a/scripts/games/grid_arena.py b/scripts/games/grid_arena.py new file mode 100644 index 0000000..03e4764 --- /dev/null +++ b/scripts/games/grid_arena.py @@ -0,0 +1,251 @@ +""" +Arena del gioco GRID TRADERS (sessione 3): 100 agenti ciechi configurano una +griglia di trading secondo STRATEGIA_GRIGLIA.md su due serie anonime +(A=BTC, B=ETH, mai rivelate; gli agenti le vedono come X/Y). Torneo standard: +3 finestre TRAIN/VALID/TEST, 90 epoche di hill-climb sul TRAIN, ogni 10 epoche +l'orchestratore blocca il 10% meno profittevole in VALID, fino a 10 superstiti. +TEST = OOS puro mai toccato dall'ottimizzazione. + + uv run python -m scripts.games.grid_arena # 100 random (smoke) + GAME_SPECS_DIR=... GAME_OUT=... uv run python -m scripts.games.grid_arena --from-specs +""" +from __future__ import annotations + +import json +import os +import random +import sys +from pathlib import Path + +from scripts.games.engine import load_anon, splits3 +from scripts.games import grid_engine +from scripts.games.grid_engine import evaluate, max_levels + +OUT = Path("data/games") +OUT.mkdir(parents=True, exist_ok=True) + +# Spazio parametri (min, max, tipo). range_*/sl_buf/tp_buf in FRAZIONE. +PSPACE = dict( + range_down=(0.02, 0.30, "f"), + range_up=(0.02, 0.30, "f"), + levels=(4, 30, "i"), + sl_buf=(0.01, 0.15, "f"), + tp_buf=(0.01, 0.15, "f"), + max_bars=(48, 3000, "i"), +) +SERIES = ["A", "B"] +TIMEFRAMES = ["15m", "30m", "1h", "2h", "4h", "1d"] # no 5m (costo computazionale) + + +def _rand(rng, lo, hi, typ): + return int(rng.randint(int(lo), int(hi))) if typ == "i" else round(rng.uniform(lo, hi), 4) + + +def random_spec(rng): + p = {k: _rand(rng, *v) for k, v in PSPACE.items()} + return {"series": rng.choice(SERIES), "tf": rng.choice(TIMEFRAMES), "params": p} + + +def _normalize(spec): + """Clampa la spec nello spazio valido e APPLICA il vincolo break-even (§4): + se il passo e' troppo fitto riduce GRID_LEVELS (come da spec: 'vanno ridotti + i GRID_LEVELS o allargato il range').""" + out = {"series": spec.get("series") if spec.get("series") in SERIES else "A", + "tf": spec.get("tf") if spec.get("tf") in TIMEFRAMES else "1h", + "params": {}} + src = spec.get("params", spec) + for k, (lo, hi, typ) in PSPACE.items(): + v = src.get(k, (lo + hi) / 2) + try: + v = float(v) + except Exception: + v = (lo + hi) / 2 + v = max(lo, min(hi, v)) + out["params"][k] = int(round(v)) if typ == "i" else round(v, 4) + p = out["params"] + lmax = max_levels(p["range_down"], p["range_up"]) + if lmax >= 2: + p["levels"] = min(p["levels"], lmax) + return out + + +def mutate(spec, rng, strength=0.25): + """Hill-climb: perturba 1-2 parametri; raramente cambia serie. + Il timeframe e' l'identita' dell'agente -> non muta.""" + s = json.loads(json.dumps(spec)) + for k in rng.sample(list(PSPACE), k=rng.randint(1, 2)): + lo, hi, typ = PSPACE[k] + span = (hi - lo) * strength + nv = max(lo, min(hi, s["params"][k] + rng.uniform(-span, span))) + s["params"][k] = int(round(nv)) if typ == "i" else round(nv, 4) + if rng.random() < 0.05: + s["series"] = rng.choice(SERIES) + return s + + +class Agent: + def __init__(self, aid, spec, brief=""): + self.id = aid + self.spec = _normalize(spec) + self.brief = brief + self.train_fit = -1e9 + self.valid_fit = -1e9 + self.metrics = {} + self.vmetrics = {} + self.alive = True + self.culled_epoch = None + + @property + def tf(self): + return self.spec["tf"] + + def score(self, datasets, sm): + data = datasets[self.tf] + tr, va, _ = sm[self.tf] + self.metrics = evaluate(data, self.spec, tr) + self.vmetrics = evaluate(data, self.spec, va) + self.train_fit = self.metrics["fitness"] + self.valid_fit = self.vmetrics["fitness"] + + +def run_tournament(specs, briefs=None, seed=2026, epochs=90, cull_every=10, + cull_n=10, out_name="grid_result.json", log=print): + rng = random.Random(seed) + used_tfs = sorted({_normalize(s)["tf"] for s in specs}) + datasets = {tf: load_anon(tf) for tf in used_tfs} + sm = {tf: splits3(datasets[tf], 0.60, 0.20) for tf in used_tfs} + briefs = briefs or [""] * len(specs) + agents = [Agent(i, s, briefs[i] if i < len(briefs) else "") + for i, s in enumerate(specs)] + for a in agents: + a.score(datasets, sm) + alive = lambda: [a for a in agents if a.alive] + log(f"[epoch 0] {len(alive())} agenti | best VALID " + f"{max(a.valid_fit for a in agents):.1f}") + history = [] + for ep in range(1, epochs + 1): + for a in alive(): + cand = _normalize(mutate(a.spec, rng)) + data = datasets[a.tf] + tr, va, _ = sm[a.tf] + m = evaluate(data, cand, tr) + if m["fitness"] > a.train_fit: + a.spec, a.metrics, a.train_fit = cand, m, m["fitness"] + a.vmetrics = evaluate(data, a.spec, va) + a.valid_fit = a.vmetrics["fitness"] + if ep % cull_every == 0: + av = sorted(alive(), key=lambda a: a.valid_fit) + k = cull_n if len(av) - cull_n >= 10 else max(0, len(av) - 10) + for a in av[:k]: + a.alive = False + a.culled_epoch = ep + log(f"[epoch {ep:2d}] cull {k:2d} -> {len(alive()):3d} vivi | " + f"best VALID {max(a.valid_fit for a in alive()):.1f} | " + f"worst-alive {min(a.valid_fit for a in alive()):.1f}") + history.append({"epoch": ep, "alive": len(alive()), + "best_valid": max(a.valid_fit for a in alive())}) + survivors = sorted(alive(), key=lambda a: a.valid_fit, reverse=True) + results = [] + for rank, a in enumerate(survivors, 1): + data = datasets[a.tf] + _, _, te = sm[a.tf] + results.append({"rank": rank, "agent": a.id, "spec": a.spec, + "brief": a.brief, "tf": a.tf, + "train": a.metrics, "valid": a.vmetrics, + "test": evaluate(data, a.spec, te), + "full": evaluate(data, a.spec, None)}) + payload = {"n_agents": len(specs), "epochs": epochs, + "survivors": len(survivors), "results": results, + "history": history, "game": "grid", + "rule": "STRATEGIA_GRIGLIA.md", + "reveal": {"A": "BTC", "B": "ETH"}} + (OUT / out_name).write_text(json.dumps(payload, indent=2)) + return payload + + +def leaderboard(payload, top=10, log=print): + log("\n========== CLASSIFICA GRID TRADERS (top %d) ==========" % top) + log("VALID = finestra del giudice | TEST = OOS puro (mai ottimizzato)") + log(f"{'#':>2} {'ag':>4} {'tf':>4} {'ser':>3} {'rng-/+':>11} {'lvl':>3} " + f"{'sl/tp buf':>11} {'mbars':>5} {'TEpnl%':>7} {'TEwin':>5} " + f"{'TEtpm':>6} {'TEsh':>5} {'VApnl%':>7}") + for r in payload["results"][:top]: + sp = r["spec"]; p = sp["params"]; te = r["test"]; va = r["valid"] + log(f"{r['rank']:>2} {r['agent']:>4} {sp['tf']:>4} {sp['series']:>3} " + f"{p['range_down']*100:>4.1f}/{p['range_up']*100:>4.1f}% {p['levels']:>3} " + f"{p['sl_buf']*100:>4.1f}/{p['tp_buf']*100:>4.1f}% {p['max_bars']:>5} " + f"{te['pnl_pct']:>7.0f} {te['win_rate']*100:>4.0f}% {te['tpm']:>6.1f} " + f"{te['sharpe']:>5.1f} {va['pnl_pct']:>7.0f}") + + +def load_specs(specs_dir, n=100): + """Carica le spec proposte dagli agenti ciechi (X->A, Y->B, pct->frazione).""" + rng = random.Random(7) + specs, briefs = [], [] + for i in range(n): + f = Path(specs_dir) / f"agent_{i}.json" + spec = None + if f.exists(): + try: + raw = json.loads(f.read_text()) + src = raw.get("params", raw) + params = { + "range_down": float(src.get("range_down_pct", src.get("range_down", 10))) , + "range_up": float(src.get("range_up_pct", src.get("range_up", 10))), + "levels": src.get("grid_levels", src.get("levels", 10)), + "sl_buf": float(src.get("sl_buf_pct", src.get("sl_buf", 5))), + "tp_buf": float(src.get("tp_buf_pct", src.get("tp_buf", 5))), + "max_bars": src.get("max_bars", 500), + } + # gli agenti parlano in percentuale -> frazione + for k in ("range_down", "range_up", "sl_buf", "tp_buf"): + if params[k] > 1.0: + params[k] = params[k] / 100.0 + spec = _normalize({ + "series": {"X": "A", "Y": "B"}.get(raw.get("series"), raw.get("series")), + "tf": raw.get("tf", "1h"), "params": params}) + briefs.append(str(raw.get("hypothesis", ""))[:300]) + except Exception: + spec = None + if spec is None: + spec = random_spec(rng) + briefs.append("(spec mancante -> sostituto casuale)") + specs.append(spec) + n_real = sum(1 for b in briefs if "mancante" not in b) + print(f"caricati {n_real}/{n} spec da agenti reali, {n - n_real} sostituiti casuali") + return specs, briefs + + +def main(): + slip = float(os.environ.get("GAME_SLIP", "0.0")) + grid_engine.set_slippage(slip) + if slip > 0: + print(f"SLIPPAGE attivo: {slip*100:.3f}%/lato") + epochs = int(os.environ.get("GAME_EPOCHS", "90")) + if "--from-specs" in sys.argv: + sd = os.environ.get("GAME_SPECS_DIR", "data/games/specs_grid") + on = os.environ.get("GAME_OUT", "grid_result.json") + specs, briefs = load_specs(sd) + payload = run_tournament(specs, briefs=briefs, epochs=epochs, out_name=on) + else: + rng = random.Random(42) + payload = run_tournament([random_spec(rng) for _ in range(100)], + seed=42, epochs=epochs) + leaderboard(payload) + rev = payload["reveal"] + w = payload["results"][0] + sp = w["spec"]; p = sp["params"] + print(f"\n>>> RIVELAZIONE: Serie X = {rev['A']}, Serie Y = {rev['B']}. " + f"Gli agenti non lo sapevano. <<<") + print(f"\nVINCITORE: agente #{w['agent']} su {sp['tf']} serie {sp['series']} | " + f"griglia -{p['range_down']*100:.1f}%/+{p['range_up']*100:.1f}% " + f"x{p['levels']} livelli, SL buf {p['sl_buf']*100:.1f}%, " + f"TP buf {p['tp_buf']*100:.1f}%, max {p['max_bars']} barre") + print(f" ipotesi dell'agente: {w['brief']}") + print(f" TEST(OOS): PnL {w['test']['pnl_pct']:.0f}% | win " + f"{w['test']['win_rate']*100:.0f}% | {w['test']['tpm']:.1f} trade/mese | " + f"Sharpe {w['test']['sharpe']:.1f}") + + +if __name__ == "__main__": + main() diff --git a/scripts/games/grid_brief.py b/scripts/games/grid_brief.py new file mode 100644 index 0000000..c3a7abe --- /dev/null +++ b/scripts/games/grid_brief.py @@ -0,0 +1,120 @@ +""" +grid_brief — digest ANONIMO per gli agenti del gioco GRID TRADERS (sessione 3). + +Come agent_brief, ma con statistiche pensate per DIMENSIONARE una griglia: +oltre a vol/autocorrelazioni, l'escursione tipica (max/min - 1) su finestre +rolling e quanto spesso il prezzo "esce" da un range simmetrico attorno a un +punto di partenza entro N barre. L'agente non sa cosa siano X e Y. + + uv run python -m scripts.games.grid_brief 1h # stampa il digest + uv run python -m scripts.games.grid_brief --all # scrive data/games/grid_digests.json +""" +from __future__ import annotations + +import json + +import numpy as np +import pandas as pd + +from scripts.games.engine import load_anon +from scripts.games.agent_brief import _stats + +TF_ID = {"15m": "T2", "30m": "T3", "1h": "T4", "2h": "T5", "4h": "T6", "1d": "T7"} + + +def _range_stats(close, windows=(100, 500, 2000)): + """Escursione (max/min - 1) su finestre rolling: mediana e p90, in %.""" + s = pd.Series(close) + out = {} + for w in windows: + if len(close) < w * 2: + continue + exc = (s.rolling(w).max() / s.rolling(w).min() - 1).dropna() + out[f"w{w}"] = {"median_pct": round(float(exc.median() * 100), 2), + "p90_pct": round(float(exc.quantile(0.9) * 100), 2)} + return out + + +def _escape_stats(close, half_widths=(0.05, 0.10, 0.20), horizon=500): + """Da un punto di partenza, % di volte in cui il prezzo esce da + +-half_width entro `horizon` barre (campionato ogni horizon/2).""" + n = len(close) + stepi = max(1, horizon // 2) + starts = np.arange(0, n - horizon, stepi) + out = {} + for hw in half_widths: + esc = 0 + for st in starts: + w = close[st:st + horizon] + p0 = w[0] + if np.any(w > p0 * (1 + hw)) or np.any(w < p0 * (1 - hw)): + esc += 1 + out[f"+-{hw*100:.0f}%"] = round(100.0 * esc / max(1, len(starts)), 1) + return out + + +def make_grid_digest(tf: str, window: int = 60): + data = load_anon(tf) + n = data["n"] + s = max(0, n - window) + dig = {"timeframe_id": TF_ID.get(tf, "T?"), "n_bars_total": n, "series": {}} + for name in ("A", "B"): + o = data[name] + c = o["close"] + norm = c[s:] / c[s] * 100.0 + dig["series"][{"A": "X", "B": "Y"}[name]] = { + "stats": _stats(c, o["high"], o["low"]), + "range_excursion_rolling": _range_stats(c), + "escape_from_range_within_500_bars_pct": _escape_stats(c), + "recent_window_norm": [round(float(v), 2) for v in norm], + } + return dig + + +GRID_MENU = { + "gioco": ("Configura una GRIGLIA di trading secondo la spec (griglia geometrica " + "FISSA dentro un range attorno al prezzo di deploy; compra quando il " + "prezzo scende attraverso un livello, rivendi quel livello quando " + "risale al livello successivo; stop-loss sotto il range e take-profit " + "sopra chiudono tutto; poi la griglia si ri-deploya sul prezzo corrente)."), + "obiettivo": ("PnL netto positivo dopo i costi (0.10% andata+ritorno per ogni " + "round-trip di livello). Servono >=10 operazioni al mese. La " + "griglia monetizza le oscillazioni e PERDE nei trend: lo stop-loss " + "limita il danno. Non sai cosa siano X e Y."), + "vincolo_break_even": ("passo_griglia = ((1+range_up)/(1-range_down))^(1/grid_levels) - 1 " + "DEVE superare 1.5 x 0.10% = 0.15%, o il bot si rifiuta " + "di partire. Griglie troppo fitte muoiono di fee."), + "parametri": { + "series": "X oppure Y", + "range_down_pct": "estremo inferiore del range, % sotto il prezzo di deploy (2-30)", + "range_up_pct": "estremo superiore del range, % sopra il prezzo di deploy (2-30)", + "grid_levels": "numero di livelli della griglia (4-30)", + "sl_buf_pct": "stop-loss: % sotto RANGE_LOW (1-15)", + "tp_buf_pct": "take-profit: % sopra RANGE_HIGH (1-15)", + "max_bars": "durata massima di una griglia in barre, poi liquida e ri-deploya (48-3000)", + }, + "trade_off": ("range stretto + tanti livelli = tanti round-trip piccoli ma SL " + "frequenti nei trend; range largo = SL rari ma capitale spesso " + "fermo. Lo stop-loss largo aumenta la perdita quando scatta; " + "stretto scatta piu' spesso. Usa le statistiche di escursione " + "del digest per dimensionare range e stop."), + "output_schema": { + "series": "X|Y", "range_down_pct": "num", "range_up_pct": "num", + "grid_levels": "int", "sl_buf_pct": "num", "tp_buf_pct": "num", + "max_bars": "int", "hypothesis": "1-2 frasi: il tuo ragionamento", + }, +} + + +if __name__ == "__main__": + import sys + from pathlib import Path + if "--all" in sys.argv: + out = {tf: make_grid_digest(tf) for tf in TF_ID} + p = Path("data/games/grid_digests.json") + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(json.dumps(out)) + print(f"scritti digest per {list(out)} -> {p}") + else: + tf = sys.argv[1] if len(sys.argv) > 1 else "1h" + print(json.dumps(make_grid_digest(tf), indent=2)[:3000]) diff --git a/scripts/games/grid_engine.py b/scripts/games/grid_engine.py new file mode 100644 index 0000000..2968e6e --- /dev/null +++ b/scripts/games/grid_engine.py @@ -0,0 +1,231 @@ +""" +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") diff --git a/scripts/games/opt_calibrate.py b/scripts/games/opt_calibrate.py new file mode 100644 index 0000000..303156a --- /dev/null +++ b/scripts/games/opt_calibrate.py @@ -0,0 +1,80 @@ +"""Calibra una superficie premi REALE dalla catena cerbero-bite -> data/games/opt_calib_*.json. + +Per ETH e BTC, dalla chain reale (OptionChain): premio mediano (ask, %spot), spread +bid/ask mediano, e IV mediana per (moneyness OTM x tenor). Piu' DVOL medio della finestra +(per scalare i premi sulla storia). + gate liquidita': max OTM con bid>0 frequente. +Cosi' il motore del gioco prezza con NUMERI REALI invece del Black-Scholes sintetico. + + uv run python -m scripts.games.opt_calibrate +""" +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.options_chain import OptionChain + +OUT = PROJECT_ROOT / "data" / "games" +# griglie: OTM firmato (put<0, call>0) e tenor in giorni +OTM_GRID = [-0.25, -0.20, -0.15, -0.10, -0.07, -0.05, -0.03, 0.0, + 0.03, 0.05, 0.07, 0.10, 0.15, 0.20, 0.25] +TEN_GRID = [7, 14, 21, 30, 45] + + +def calibrate(asset: str): + oc = OptionChain(asset) + d = oc.df.copy() + spot = oc._spot_proxy() + d["spot"] = d["timestamp"].map(spot) + d = d.dropna(subset=["spot", "ask", "bid", "iv"]) + d = d[d["ask"] > 0] + d["otm"] = d["strike"] / d["spot"] - 1.0 # firmato: <0 put OTM, >0 call OTM + d["prem_pct"] = d["ask"] * 100.0 # ask in coin -> %notional + d["spread"] = (d["ask"] - d["bid"]) / ((d["ask"] + d["bid"]) / 2) + d["sellable"] = (d["bid"] > 0).astype(float) + # superficie: per ciascun (tipo, otm_bin, tenor_bin) -> mediane + surf = {"P": {}, "C": {}} + for typ in ("P", "C"): + dt = d[d["option_type"] == typ] + for ten in TEN_GRID: + tlo, thi = ten * 0.6, ten * 1.6 + dtt = dt[(dt["tenor_d"] >= tlo) & (dt["tenor_d"] <= thi)] + for otm in OTM_GRID: + # banda moneyness +-1.5% attorno al target + band = dtt[(dtt["otm"] >= otm - 0.02) & (dtt["otm"] <= otm + 0.02)] + if len(band) < 5: + continue + surf[typ][f"{otm:+.2f}|{ten}"] = dict( + prem=round(float(band["prem_pct"].median()), 4), + spread=round(float(band["spread"].median()), 4), + iv=round(float(band["iv"].median()), 4), + sellable=round(float(band["sellable"].mean()), 3), + n=int(len(band))) + dvol_avg = float(np.nanmedian(d["iv"][d["otm"].abs() < 0.03])) # ~ATM IV medio + # gate liquidita': OTM piu' profondo (put) con bid>0 nel >=50% dei casi + puts = d[d["option_type"] == "P"] + deep = puts[puts["otm"] <= -0.10] + out = {"asset": asset, "dvol_chain": round(dvol_avg, 4), + "surface": surf, "otm_grid": OTM_GRID, "ten_grid": TEN_GRID, + "window": [str(oc.df["ts"].min())[:10], str(oc.df["ts"].max())[:10]]} + (OUT / f"opt_calib_{asset.lower()}.json").write_text(json.dumps(out)) + npts = len(surf["P"]) + len(surf["C"]) + print(f"{asset}: {npts} punti superficie | ATM IV ~{dvol_avg:.2f} | finestra {out['window']}") + # stampa qualche premio reale put per sanity + for key in ["-0.05|14", "-0.10|14", "-0.15|30", "-0.20|45"]: + v = surf["P"].get(key) + if v: + print(f" put {key:>9}: prem {v['prem']:.2f}% spread {v['spread']*100:.0f}% " + f"iv {v['iv']:.0f}% sellable {v['sellable']*100:.0f}% (n={v['n']})") + + +if __name__ == "__main__": + for a in ("BTC", "ETH"): + calibrate(a) diff --git a/scripts/games/options_arena.py b/scripts/games/options_arena.py new file mode 100644 index 0000000..009cb32 --- /dev/null +++ b/scripts/games/options_arena.py @@ -0,0 +1,190 @@ +""" +Arena del gioco-OPZIONI: 100 agenti ciechi propongono STRUTTURE in opzioni su due +serie anonime (A=BTC, B=ETH). Torneo identico al gioco-prezzi (3 finestre TRAIN/VALID/ +TEST, 90 epoche, cull 10% ogni 10 epoche -> 10 finalisti), ma le strategie sono opzioni +prezzate con BS + skew + DVOL (scripts/games/options_engine.py). + + uv run python -m scripts.games.options_arena # 100 agenti random (test) + GAME_SPECS_DIR=... GAME_OUT=... uv run python -m scripts.games.options_arena --from-specs +""" +from __future__ import annotations + +import json +import os +import random +import sys +from pathlib import Path + +import numpy as np + +from scripts.games.options_engine import (load_opt, splits3, evaluate, STRUCTURES) + +OUT = Path("data/games"); OUT.mkdir(parents=True, exist_ok=True) + +# spazio parametri: (min, max, tipo) +PSPACE = dict(otm=(0.02, 0.20, "f"), width=(0.02, 0.12, "f"), dte=(7, 45, "i")) +SERIES = ["A", "B"] + + +def _rand(rng, lo, hi, typ): + return int(rng.randint(int(lo), int(hi))) if typ == "i" else round(rng.uniform(lo, hi), 3) + + +def random_spec(rng): + p = {k: _rand(rng, *v) for k, v in PSPACE.items()} + return {"structure": rng.choice(STRUCTURES), "series": rng.choice(SERIES), "params": p} + + +def _normalize(spec): + st = spec.get("structure") + if st not in STRUCTURES: + st = "short_put" + out = {"structure": st, "series": spec.get("series") if spec.get("series") in SERIES else "A", + "params": {}} + src = spec.get("params", {}) + for k, (lo, hi, typ) in PSPACE.items(): + v = src.get(k, (lo + hi) / 2) + try: + v = float(v) + except Exception: + v = (lo + hi) / 2 + v = max(lo, min(hi, v)) + out["params"][k] = int(round(v)) if typ == "i" else round(v, 3) + # flatten per evaluate (structure/otm/width/dte) + out["structure"] = st + return out + + +def _flat(spec): + return {"structure": spec["structure"], **spec["params"]} + + +def mutate(spec, rng, strength=0.25): + s = json.loads(json.dumps(spec)) + keys = list(PSPACE) + for k in rng.sample(keys, k=rng.randint(1, 2)): + lo, hi, typ = PSPACE[k] + span = (hi - lo) * strength + nv = max(lo, min(hi, s["params"][k] + rng.uniform(-span, span))) + s["params"][k] = int(round(nv)) if typ == "i" else round(nv, 3) + if rng.random() < 0.12: + s["structure"] = rng.choice(STRUCTURES) + if rng.random() < 0.05: + s["series"] = rng.choice(SERIES) + return s + + +class Agent: + def __init__(self, aid, spec, brief=""): + self.id = aid + self.spec = _normalize(spec) + self.brief = brief + self.train_fit = self.valid_fit = -1e9 + self.metrics = self.vmetrics = {} + self.alive = True + + @property + def series(self): + return self.spec["series"] + + def score(self, datasets, splits_map): + d = datasets[self.series]; tr, va, _ = splits_map[self.series] + self.metrics = evaluate(d, _flat(self.spec), tr) + self.vmetrics = evaluate(d, _flat(self.spec), va) + self.train_fit = self.metrics["fitness"]; self.valid_fit = self.vmetrics["fitness"] + + +def run_tournament(specs, briefs=None, seed=2026, epochs=90, cull_every=10, cull_n=10, + out_name="options_result.json", log=print): + rng = random.Random(seed) + datasets = {"A": load_opt("BTC"), "B": load_opt("ETH")} + splits_map = {k: splits3(datasets[k]) for k in datasets} + briefs = briefs or [""] * len(specs) + agents = [Agent(i, s, briefs[i] if i < len(briefs) else "") for i, s in enumerate(specs)] + for a in agents: + a.score(datasets, splits_map) + alive = lambda: [a for a in agents if a.alive] + log(f"[epoch 0] {len(alive())} agenti | best VALID {max(a.valid_fit for a in agents):.1f}") + for ep in range(1, epochs + 1): + for a in alive(): + cand = _normalize(mutate(a.spec, rng)) + d = datasets[cand["series"]]; tr, va, _ = splits_map[cand["series"]] + m = evaluate(d, _flat(cand), tr) + if m["fitness"] > a.train_fit: + a.spec, a.metrics, a.train_fit = cand, m, m["fitness"] + a.vmetrics = evaluate(d, _flat(cand), va); a.valid_fit = a.vmetrics["fitness"] + if ep % cull_every == 0: + av = sorted(alive(), key=lambda a: a.valid_fit) + k = cull_n if len(av) - cull_n >= 10 else max(0, len(av) - 10) + for a in av[:k]: + a.alive = False + log(f"[epoch {ep:2d}] cull {k:2d} -> {len(alive()):3d} | best VALID " + f"{max(a.valid_fit for a in alive()):.1f} | worst {min(a.valid_fit for a in alive()):.1f}") + survivors = sorted(alive(), key=lambda a: a.valid_fit, reverse=True) + results = [] + for rank, a in enumerate(survivors, 1): + d = datasets[a.series]; _, _, te = splits_map[a.series] + results.append({"rank": rank, "agent": a.id, "spec": a.spec, "brief": a.brief, + "series": a.series, "train": a.metrics, "valid": a.vmetrics, + "test": evaluate(d, _flat(a.spec), te), "full": evaluate(d, _flat(a.spec), None)}) + payload = {"n_agents": len(specs), "survivors": len(survivors), "results": results, + "reveal": {"A": "BTC", "B": "ETH"}, "game": "options"} + (OUT / out_name).write_text(json.dumps(payload, indent=2)) + return payload + + +def leaderboard(payload, top=10, log=print): + log("\n========= CLASSIFICA FINALE OPZIONI (top %d) =========" % top) + log(f"{'#':>2} {'ag':>4} {'ser':>3} {'struttura':>14} {'otm':>5} {'dte':>4} " + f"{'TEpnl%':>8} {'TEwin':>5} {'TEtpm':>6} {'TEsh':>6}") + for r in payload["results"][:top]: + sp = r["spec"]; te = r["test"]; p = sp["params"] + log(f"{r['rank']:>2} {r['agent']:>4} {sp['series']:>3} {sp['structure']:>14} " + f"{p['otm']:>5.2f} {p['dte']:>4} {te['pnl_pct']:>8.0f} {te['win_rate']*100:>4.0f}% " + f"{te['tpm']:>6.0f} {te['sharpe']:>6.1f}") + + +def load_specs(specs_dir, n=100): + rng = random.Random(7); specs, briefs = [], [] + for i in range(n): + f = Path(specs_dir) / f"agent_{i}.json" + spec = None + if f.exists(): + try: + raw = json.loads(f.read_text()) + params = {k: raw.get(k, raw.get("params", {}).get(k)) for k in PSPACE} + spec = _normalize({"structure": raw.get("structure"), + "series": {"X": "A", "Y": "B"}.get(raw.get("series"), raw.get("series")), + "params": params}) + briefs.append(str(raw.get("hypothesis", ""))[:300]) + except Exception: + spec = None + if spec is None: + spec = random_spec(rng); briefs.append("(spec mancante -> random)") + specs.append(spec) + return specs, briefs + + +def main(): + if "--from-specs" in sys.argv: + sd = os.environ.get("GAME_SPECS_DIR", "data/games/specs_opt") + on = os.environ.get("GAME_OUT", "options_result.json") + specs, briefs = load_specs(sd) + n_real = sum(1 for b in briefs if "mancante" not in b) + print(f"caricati {n_real}/100 spec da agenti reali") + payload = run_tournament(specs, briefs=briefs, out_name=on) + else: + rng = random.Random(42) + payload = run_tournament([random_spec(rng) for _ in range(100)], seed=42) + leaderboard(payload) + rev = payload["reveal"]; w = payload["results"][0] + print(f"\n>>> RIVELAZIONE: A={rev['A']}, B={rev['B']}. Gli agenti non lo sapevano. <<<") + print(f"VINCITORE: #{w['agent']} {w['series']} {w['spec']['structure']} " + f"otm{w['spec']['params']['otm']} dte{w['spec']['params']['dte']}") + print(f" ipotesi: {w['brief']}") + print(f" TEST: PnL {w['test']['pnl_pct']:.0f}% | win {w['test']['win_rate']*100:.0f}% | " + f"{w['test']['tpm']:.0f} tr/mese | Sharpe {w['test']['sharpe']:.1f}") + + +if __name__ == "__main__": + main() diff --git a/scripts/games/options_engine.py b/scripts/games/options_engine.py new file mode 100644 index 0000000..cfc11d2 --- /dev/null +++ b/scripts/games/options_engine.py @@ -0,0 +1,215 @@ +""" +Motore del gioco-OPZIONI: prezza e backtesta strutture in opzioni proposte dagli +agenti ciechi, sui prezzi REALI ETH/BTC, con Black-Scholes + skew fittato + DVOL storica. + +NON usa la chain reale (solo 6 settimane, un regime): prezza sinteticamente con la +vol implicita storica (DVOL Deribit, dal 2021-03) e la curva di skew fittata sulle IV +reali della ricerca credit-spread (iv/atm = 1 - 0.664*k + 3.494*k^2, k=ln(K/S)). Costi: +haircut bid/ask sulle opzioni (il fill reale e' peggiore del mid). Roll giornaliero, +hold-to-expiry (terminale model-free dai prezzi reali). PnL per-trade ADDITIVO. + +Caveat onesto (dalla ricerca del progetto): il premium-selling a skew negativo vince nei +campioni calmi e restituisce tutto nei crash -> il gioco lo mostrera'. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +import json as _json + +from src.data.downloader import load_data +from scripts.analysis.option_overlay_lab import bs_put, bs_call, dvol_for + +# skew fittato (fallback se manca la calibrazione reale): iv/atm in funzione di k=ln(K/S). +SKEW_A, SKEW_B = -0.664, 3.494 +MIN_TRADES_PER_MONTH = 10.0 +TRADING_DAYS_MONTH = 30.0 + +# --- pricing REALE: superficie premi/spread da cerbero-bite (scripts/games/opt_calibrate.py) --- +_CALIB_DIR = PROJECT_ROOT / "data" / "games" +_CALIB = {} + + +def _load_calib(asset): + if asset not in _CALIB: + f = _CALIB_DIR / f"opt_calib_{asset.lower()}.json" + _CALIB[asset] = _json.loads(f.read_text()) if f.exists() else None + return _CALIB[asset] + + +def _surf_lookup(cal, typ, otm_signed, dte): + """Premio% e spread reali per (otm firmato, dte): punto di griglia piu' vicino. + Ritorna (prem_pct, spread, sellable) o None se fuori dalla superficie liquida.""" + s = cal["surface"][typ] + og = cal["otm_grid"]; tg = cal["ten_grid"] + o = min(og, key=lambda x: abs(x - otm_signed)) + t = min(tg, key=lambda x: abs(x - dte)) + if abs(o - otm_signed) > 0.06: # troppo lontano dagli strike reali -> illiquido + return None + v = s.get(f"{o:+.2f}|{t}") + if not v or v["sellable"] < 0.5: + return None + return v["prem"], v["spread"], v["sellable"] + + +def iv_skew(k: float, atm: float) -> float: + """IV per moneyness k=ln(K/S) dato l'ATM vol. Clamp a [0.3x, 3x] atm.""" + mult = 1.0 + SKEW_A * k + SKEW_B * k * k + mult = min(max(mult, 0.3), 3.0) + return atm * mult + + +def load_opt(asset: str = "ETH"): + """Prezzi GIORNALIERI (resample 1h->1d) + DVOL allineata. asset reale nascosto.""" + df = load_data(asset, "1h").copy() + df["dt"] = pd.to_datetime(df["datetime"]) + g = df.set_index("dt").resample("1D").agg( + {"timestamp": "first", "open": "first", "high": "max", "low": "min", + "close": "last"}).dropna(subset=["close"]).reset_index(drop=True) + g["timestamp"] = g["timestamp"].astype("int64") + dv = dvol_for(g, asset) + cal = _load_calib(asset) + dvol_chain = (cal["dvol_chain"] / 100.0) if cal else float(np.nanmedian(dv)) + return {"close": g["close"].to_numpy(float), "high": g["high"].to_numpy(float), + "low": g["low"].to_numpy(float), "dvol": dv, "asset": asset, + "dvol_chain": dvol_chain, "real": cal is not None, + "dt": pd.to_datetime(g["timestamp"], unit="ms", utc=True).to_numpy(), + "n": len(g)} + + +# -------------------------------------------------------------------------- +# Pricing di una struttura: ritorna (premio_netto_incassato, funzione_payoff(ST)) +# premio>0 = struttura a CREDITO (vendi); payoff e' il valore terminale (>=0 per long opt). +# Convenzione PnL trade: net = (premio_incassato - payoff_terminale)/S0 - costi (per credito) +# Tutto normalizzato sul SPOT (frazione), cosi' e' confrontabile fra asset/epoche. +# -------------------------------------------------------------------------- +STRUCTURES = ["short_put", "short_call", "short_strangle", "put_spread", + "call_spread", "iron_condor", "long_put", "long_call", "long_straddle"] + + +def _legs_for(struct, S, otm, width): + kp = S * (1 - otm); kc = S * (1 + otm) + kp2 = S * (1 - otm - width); kc2 = S * (1 + otm + width) + return { + "short_put": [("P", kp, -1)], "short_call": [("C", kc, -1)], + "short_strangle": [("P", kp, -1), ("C", kc, -1)], + "put_spread": [("P", kp, -1), ("P", kp2, +1)], + "call_spread": [("C", kc, -1), ("C", kc2, +1)], + "iron_condor": [("P", kp, -1), ("P", kp2, +1), ("C", kc, -1), ("C", kc2, +1)], + "long_put": [("P", kp, +1)], "long_call": [("C", kc, +1)], + "long_straddle": [("P", S, +1), ("C", S, +1)], + }[struct] + + +def _price_real(struct, S, dte, scale, otm, width, cal): + """Pricing REALE dalla superficie cerbero-bite. Ritorna (entry_cf_frac, legs, ok). + entry_cf_frac = cassa d'ingresso in frazione di spot (>0 = incassi); side-aware bid/ask; + ok=False se una gamba e' fuori dagli strike liquidi reali.""" + legs = _legs_for(struct, S, otm, width) + entry = 0.0 + for typ, K, sgn in legs: + q = _surf_lookup(cal, typ, K / S - 1.0, dte) + if q is None: + return 0.0, legs, False + prem, spread, _ = q + pf = prem / 100.0 * scale # premio frazione di spot, scalato a DVOL del giorno + if sgn < 0: # short: incassi il BID (~ ask*(1-spread)) + entry += pf * (1 - spread) + else: # long: paghi l'ASK + entry -= pf + return entry, legs, True + + +def _price(struct, S, T, atm, otm, width): + """Fallback SINTETICO (BS+skew). Usato solo se manca la calibrazione reale.""" + legs = _legs_for(struct, S, otm, width) + prem = gross = 0.0 + for typ, K, sgn in legs: + px = bs_put(S, K, T, iv_skew(np.log(K / S), atm)) if typ == "P" \ + else bs_call(S, K, T, iv_skew(np.log(K / S), atm)) + prem += -sgn * px / S + gross += abs(px) / S + return prem - 0.06 * gross, legs, True + + +def _payoff(legs, ST): + v = 0.0 + for typ, K, sgn in legs: + intr = max(K - ST, 0.0) if typ == "P" else max(ST - K, 0.0) + v += sgn * intr # valore terminale delle opzioni che POSSIEDI/devi + return v # per le short questo e' cio' che PAGHI (sgn<0 -> negativo = debito) + + +def evaluate(data, spec, sl=None): + """Backtest della struttura: roll giornaliero, hold dte giorni, PnL additivo. + spec = {structure, otm, width, dte}. Ritorna metriche con scoring PNL+%win, >=10 tr/mese. + """ + c, dv = data["close"], data["dvol"] + n = data["n"] + s, e = (sl if sl else (0, n)) + struct = spec["structure"] + otm = float(spec["otm"]); width = float(spec.get("width", 0.05)) + dte = int(spec["dte"]) + T = dte / 365.0 + cal = _load_calib(data["asset"]); dvol_chain = data["dvol_chain"] + rets = [] + i = s + while i < e - dte: + S0 = c[i]; atm = dv[i] + if S0 <= 0 or atm <= 0: + i += 1; continue + if cal is not None: # PRICING REALE (cerbero-bite), scalato a DVOL del giorno + scale = min(max(atm / dvol_chain, 0.3), 4.0) + entry, legs, ok = _price_real(struct, S0, dte, scale, otm, width, cal) + if not ok: # strike fuori dalla superficie liquida reale -> non eseguibile + i += 1; continue + net = entry + _payoff(legs, c[i + dte]) / S0 + else: # fallback sintetico + prem, legs, _ = _price(struct, S0, T, atm, otm, width) + net = prem + _payoff(legs, c[i + dte]) / S0 + rets.append(net) + i += 1 # roll giornaliero (posizioni sovrapposte) + rets = np.array(rets) + nbars = e - s + months = nbars / TRADING_DAYS_MONTH + 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, fitness=-1e6) + win = 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 + if not qualified: + fitness = -1e6 + pnl + return dict(n_trades=n_tr, win_rate=win, pnl_pct=pnl, tpm=tpm, sharpe=sharpe, + avg_ret=avg, qualified=qualified, fitness=fitness) + + +def splits3(data, train_frac=0.60, valid_frac=0.20): + n = data["n"] + c1 = int(n * train_frac); c2 = int(n * (train_frac + valid_frac)) + return (0, c1), (c1, c2), (c2, n) + + +if __name__ == "__main__": + d = load_opt("ETH") + print("loaded", d["n"], "giorni", str(d["dt"][0])[:10], "->", str(d["dt"][-1])[:10], + "| dvol", round(float(np.nanmean(d["dvol"])), 2)) + tr, va, te = splits3(d) + for st in ["short_put", "short_strangle", "iron_condor", "long_straddle", "put_spread"]: + sp = {"structure": st, "otm": 0.05, "width": 0.05, "dte": 14} + f = evaluate(d, sp, None); o = evaluate(d, sp, te) + print(f"{st:14} FULL pnl{f['pnl_pct']:8.0f} win{f['win_rate']*100:4.0f} tpm{f['tpm']:5.0f} " + f"Sh{f['sharpe']:6.1f} | OOS pnl{o['pnl_pct']:8.0f} win{o['win_rate']*100:4.0f} Sh{o['sharpe']:6.1f}") diff --git a/scripts/games/run_game.py b/scripts/games/run_game.py index df2f79e..7042a54 100644 --- a/scripts/games/run_game.py +++ b/scripts/games/run_game.py @@ -14,6 +14,7 @@ import random from pathlib import Path from scripts.games import engine +from scripts.games import arena from scripts.games.arena import random_spec, run_tournament, leaderboard, _normalize SPECS_DIR = Path(os.environ.get("GAME_SPECS_DIR", "data/games/specs")) @@ -59,6 +60,9 @@ def load_specs(): def main(): slip = float(os.environ.get("GAME_SLIP", "0.0")) engine.set_slippage(slip) + if os.environ.get("GAME_NO_LIVE") == "1": + arena.set_no_live(True) + print("VINCOLO: solo strategie NON in live (no pairs, no zscore/breakout-reversion)") if slip > 0: print(f"SLIPPAGE attivo: {slip*100:.3f}%/lato " f"(single-leg {2*slip*100:.2f}% RT extra, pairs {4*slip*100:.2f}% extra)") diff --git a/scripts/games/session_arena.py b/scripts/games/session_arena.py new file mode 100644 index 0000000..527d120 --- /dev/null +++ b/scripts/games/session_arena.py @@ -0,0 +1,174 @@ +""" +Arena del gioco-SESSION: 100 agenti ciechi cercano pattern ORARI intraday (fascia di +controllo -> finestra successiva) su due serie anonime (A=BTC, B=ETH). Torneo standard +(3 finestre, 90 epoche, cull 10%/10) col motore session_engine. + + uv run python -m scripts.games.session_arena # 100 random (test) + GAME_SPECS_DIR=... GAME_OUT=... uv run python -m scripts.games.session_arena --from-specs +""" +from __future__ import annotations + +import json +import os +import random +import sys +from pathlib import Path + +from scripts.games.session_engine import load_session, splits3, evaluate + +OUT = Path("data/games"); OUT.mkdir(parents=True, exist_ok=True) +PSPACE = dict(ctrl_hour=(0, 23, "i"), ctrl_len=(1, 6, "i"), + entry_thr=(0.0, 1.5, "f"), hold=(1, 12, "i")) +SERIES = ["A", "B"] +DIRECTIONS = ["trend", "reversion"] + + +def _rand(rng, lo, hi, typ): + return int(rng.randint(int(lo), int(hi))) if typ == "i" else round(rng.uniform(lo, hi), 3) + + +def random_spec(rng): + p = {k: _rand(rng, *v) for k, v in PSPACE.items()} + return {"series": rng.choice(SERIES), "direction": rng.choice(DIRECTIONS), "params": p} + + +def _normalize(spec): + out = {"series": spec.get("series") if spec.get("series") in SERIES else "A", + "direction": spec.get("direction") if spec.get("direction") in DIRECTIONS else "trend", + "params": {}} + src = spec.get("params", spec) + for k, (lo, hi, typ) in PSPACE.items(): + v = src.get(k, (lo + hi) / 2) + try: + v = float(v) + except Exception: + v = (lo + hi) / 2 + v = max(lo, min(hi, v)) + out["params"][k] = int(round(v)) if typ == "i" else round(v, 3) + return out + + +def _flat(spec): + return {"direction": spec["direction"], **spec["params"]} + + +def mutate(spec, rng, strength=0.25): + s = json.loads(json.dumps(spec)) + for k in rng.sample(list(PSPACE), k=rng.randint(1, 2)): + lo, hi, typ = PSPACE[k] + span = (hi - lo) * strength + nv = max(lo, min(hi, s["params"][k] + rng.uniform(-span, span))) + s["params"][k] = int(round(nv)) if typ == "i" else round(nv, 3) + if rng.random() < 0.12: + s["direction"] = rng.choice(DIRECTIONS) + if rng.random() < 0.05: + s["series"] = rng.choice(SERIES) + return s + + +class Agent: + def __init__(self, aid, spec, brief=""): + self.id = aid; self.spec = _normalize(spec); self.brief = brief + self.train_fit = self.valid_fit = -1e9; self.metrics = self.vmetrics = {}; self.alive = True + + @property + def series(self): + return self.spec["series"] + + def score(self, datasets, sm): + d = datasets[self.series]; tr, va, _ = sm[self.series] + self.metrics = evaluate(d, _flat(self.spec), tr); self.vmetrics = evaluate(d, _flat(self.spec), va) + self.train_fit = self.metrics["fitness"]; self.valid_fit = self.vmetrics["fitness"] + + +def run_tournament(specs, briefs=None, seed=2026, epochs=90, cull_every=10, cull_n=10, + out_name="session_result.json", log=print): + rng = random.Random(seed) + datasets = {"A": load_session("BTC"), "B": load_session("ETH")} + sm = {k: splits3(datasets[k]) for k in datasets} + briefs = briefs or [""] * len(specs) + agents = [Agent(i, s, briefs[i] if i < len(briefs) else "") for i, s in enumerate(specs)] + for a in agents: + a.score(datasets, sm) + alive = lambda: [a for a in agents if a.alive] + log(f"[epoch 0] {len(alive())} | best VALID {max(a.valid_fit for a in agents):.1f}") + for ep in range(1, epochs + 1): + for a in alive(): + cand = _normalize(mutate(a.spec, rng)) + d = datasets[cand["series"]]; tr, va, _ = sm[cand["series"]] + m = evaluate(d, _flat(cand), tr) + if m["fitness"] > a.train_fit: + a.spec, a.metrics, a.train_fit = cand, m, m["fitness"] + a.vmetrics = evaluate(d, _flat(cand), va); a.valid_fit = a.vmetrics["fitness"] + if ep % cull_every == 0: + av = sorted(alive(), key=lambda a: a.valid_fit) + k = cull_n if len(av) - cull_n >= 10 else max(0, len(av) - 10) + for a in av[:k]: + a.alive = False + log(f"[epoch {ep:2d}] cull {k:2d} -> {len(alive()):3d} | best VALID " + f"{max(a.valid_fit for a in alive()):.1f} | worst {min(a.valid_fit for a in alive()):.1f}") + survivors = sorted(alive(), key=lambda a: a.valid_fit, reverse=True) + results = [] + for rank, a in enumerate(survivors, 1): + d = datasets[a.series]; _, _, te = sm[a.series] + results.append({"rank": rank, "agent": a.id, "spec": a.spec, "brief": a.brief, + "series": a.series, "train": a.metrics, "valid": a.vmetrics, + "test": evaluate(d, _flat(a.spec), te), "full": evaluate(d, _flat(a.spec), None)}) + payload = {"n_agents": len(specs), "survivors": len(survivors), "results": results, + "reveal": {"A": "BTC", "B": "ETH"}, "game": "session"} + (OUT / out_name).write_text(json.dumps(payload, indent=2)) + return payload + + +def leaderboard(payload, top=10, log=print): + log("\n===== CLASSIFICA SESSION (top %d) — fascia controllo -> finestra dopo =====" % top) + log(f"{'#':>2} {'ag':>4} {'ser':>3} {'h':>3} {'len':>3} {'thr%':>5} {'hold':>4} {'dir':>9} " + f"{'TEpnl%':>8} {'TEwin':>5} {'TEtpm':>6} {'TEsh':>6}") + for r in payload["results"][:top]: + sp = r["spec"]; te = r["test"]; p = sp["params"] + log(f"{r['rank']:>2} {r['agent']:>4} {sp['series']:>3} {p['ctrl_hour']:>3} {p['ctrl_len']:>3} " + f"{p['entry_thr']:>5.2f} {p['hold']:>4} {sp['direction']:>9} {te['pnl_pct']:>8.0f} " + f"{te['win_rate']*100:>4.0f}% {te['tpm']:>6.0f} {te['sharpe']:>6.1f}") + + +def load_specs(specs_dir, n=100): + rng = random.Random(7); specs, briefs = [], [] + for i in range(n): + f = Path(specs_dir) / f"agent_{i}.json"; spec = None + if f.exists(): + try: + raw = json.loads(f.read_text()) + params = {k: raw.get(k, raw.get("params", {}).get(k)) for k in PSPACE} + spec = _normalize({"series": {"X": "A", "Y": "B"}.get(raw.get("series"), raw.get("series")), + "direction": raw.get("direction"), "params": params}) + briefs.append(str(raw.get("hypothesis", ""))[:300]) + except Exception: + spec = None + if spec is None: + spec = random_spec(rng); briefs.append("(spec mancante -> random)") + specs.append(spec) + return specs, briefs + + +def main(): + if "--from-specs" in sys.argv: + sd = os.environ.get("GAME_SPECS_DIR", "data/games/specs_sess") + on = os.environ.get("GAME_OUT", "session_result.json") + specs, briefs = load_specs(sd) + print(f"caricati {sum(1 for b in briefs if 'mancante' not in b)}/100 spec da agenti reali") + payload = run_tournament(specs, briefs=briefs, out_name=on) + else: + rng = random.Random(42) + payload = run_tournament([random_spec(rng) for _ in range(100)], seed=42) + leaderboard(payload) + rev = payload["reveal"]; w = payload["results"][0]; p = w["spec"]["params"] + print(f"\n>>> RIVELAZIONE: A={rev['A']}, B={rev['B']}. <<<") + print(f"VINCITORE: #{w['agent']} {w['series']} fascia h{p['ctrl_hour']} len{p['ctrl_len']} " + f"-> {w['spec']['direction']} hold{p['hold']}h thr{p['entry_thr']}%") + print(f" ipotesi: {w['brief']}") + print(f" TEST: PnL {w['test']['pnl_pct']:.0f}% | win {w['test']['win_rate']*100:.0f}% | " + f"{w['test']['tpm']:.0f} tr/mese | Sharpe {w['test']['sharpe']:.1f}") + + +if __name__ == "__main__": + main() diff --git a/scripts/games/session_engine.py b/scripts/games/session_engine.py new file mode 100644 index 0000000..fb30a4e --- /dev/null +++ b/scripts/games/session_engine.py @@ -0,0 +1,98 @@ +""" +Motore del gioco-SESSION: pattern ORARI intraday. Ogni giorno si osserva il movimento +in una "fascia di controllo" [ctrl_hour, ctrl_hour+ctrl_len) e si scommette sul movimento +della finestra SUBITO DOPO (hold ore), seguendo (trend) o fadando (reversion) la fascia. + +Cerca se esistono orari il cui comportamento ANTICIPA la finestra successiva, ripetibile nei +giorni. Dati orari reali (BTC=A, ETH=B), full history. PnL per-trade additivo, fee 0.10% RT. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from src.data.downloader import load_data + +FEE_RT = 0.001 +MIN_TRADES_PER_MONTH = 10.0 +BARS_PER_MONTH = 24 * 30 + + +def load_session(asset: str = "BTC"): + df = load_data(asset, "1h").copy() + dt = pd.to_datetime(df["datetime"]) + return {"close": df["close"].to_numpy(float), + "open": df["open"].to_numpy(float), + "hour": dt.dt.hour.to_numpy(), + "day": (dt.dt.year * 366 + dt.dt.dayofyear).to_numpy(), # indice giorno + "dt": dt.to_numpy(), "n": len(df)} + + +def evaluate(data, spec, sl=None, fee=FEE_RT): + """spec = {ctrl_hour, ctrl_len, entry_thr(%), direction, hold}. Una valutazione per giorno: + a fine fascia di controllo, se |ret_fascia| > entry_thr entra e tiene hold ore.""" + c, hour = data["close"], data["hour"] + n = data["n"] + s, e = (sl if sl else (0, n)) + ch = int(spec["ctrl_hour"]) % 24 + cl = max(1, int(spec["ctrl_len"])) + thr = float(spec["entry_thr"]) / 100.0 + hold = max(1, int(spec["hold"])) + sign = 1 if spec.get("direction", "trend") == "trend" else -1 + + # indici in cui inizia la fascia di controllo (bar all'ora ch) + starts = np.where(hour[s:e] == ch)[0] + s + rets = [] + for st in starts: + be = st + cl - 1 # ultima barra della fascia + ex = be + hold # uscita + if ex >= e or st == 0: + continue + ctrl_ret = c[be] / c[st - 1] - 1.0 # ritorno della fascia (causale: chiude a be) + if abs(ctrl_ret) < thr: + continue + d = sign * (1 if ctrl_ret > 0 else -1) # trend segue, reversion fada + entry = c[be]; exit_px = c[ex] + net = d * (exit_px - entry) / entry - fee + rets.append(net) + rets = np.array(rets) + nbars = e - s + months = nbars / BARS_PER_MONTH + 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, fitness=-1e6) + win = 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 + if not qualified: + fitness = -1e6 + pnl + return dict(n_trades=n_tr, win_rate=win, pnl_pct=pnl, tpm=tpm, sharpe=sharpe, + avg_ret=avg, qualified=qualified, fitness=fitness) + + +def splits3(data, train_frac=0.60, valid_frac=0.20): + n = data["n"] + c1 = int(n * train_frac); c2 = int(n * (train_frac + valid_frac)) + return (0, c1), (c1, c2), (c2, n) + + +if __name__ == "__main__": + d = load_session("BTC"); tr, va, te = splits3(d) + for ch in [0, 8, 13, 20]: + for dr in ["trend", "reversion"]: + sp = {"ctrl_hour": ch, "ctrl_len": 2, "entry_thr": 0.3, "direction": dr, "hold": 4} + f = evaluate(d, sp, None); o = evaluate(d, sp, te) + print(f"h{ch:>2} {dr:>9} len2 hold4 thr0.3 | FULL pnl{f['pnl_pct']:7.0f} win{f['win_rate']*100:3.0f} " + f"tpm{f['tpm']:4.0f} Sh{f['sharpe']:5.1f} | OOS Sh{o['sharpe']:5.1f}")