Adriano Dal Pastro
14522262e6
Reset del progetto su fondamenta verificate dopo la scoperta che l'intera libreria "validata OOS" era artefatto di feed contaminato (print fantasma del feed Cerbero TESTNET + storico Binance/USDT). - Storico ricostruito da Deribit MAINNET (ccxt pubblico, tokenless) e CERTIFICATO (certify_feed.py): BTC/ETH puliti su TUTTA la storia (mediana 2-6 bps vs Coinbase USD), integrita' OHLC + coerenza resample (maxΔ 0.00) + cross-venue OK. Alt esclusi (illiquidi/divergenti: LTC/DOGE 50-82% barre flat; XRP/BNB non certificabili). - Verdetto sul feed pulito: FADE / PAIRS / XS01 / TSM01 morti (ogni portafoglio Sharpe -2.3..-3.0, DD ~40%); solo SH01 e frammenti HONEST con segnale residuo, da ri-validare in isolamento. - Cleanup "restart pulito": strategie, stack live (src/live, src/portfolio, runner/executor, yml, docker), ~100 script ricerca/gate, waste/games/ portfolios, dati non certificati + cache e 60+ diari -> archiviati in Old/ (preservati, non cancellati). Diario consolidato in un unico documento. - Skeleton ricerca tenuto: Strategy ABC + indicatori + src/fractal + src/backtest/engine + load_data; tool dati certificati (rebuild_history, certify_feed, audit_feed, multi_source_check). - Universo dati ATTIVO: solo BTC/ETH (5m/15m/1h); guardrail fisico (load_data su alt -> FileNotFoundError). Esecuzione DISABILITATA, conto flat. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
254 lines
10 KiB
Python
254 lines
10 KiB
Python
"""
|
|
Arena — tournament orchestrator per il gioco "Blind Traders".
|
|
|
|
100 agenti partono da una spec di strategia (creata alla cieca: vedi
|
|
agent_brief.py / workflow). L'orchestratore valuta ogni spec con il backtest
|
|
deterministico (engine.evaluate) su TRAIN, da' epoche di elaborazione (ogni
|
|
agente affina la propria strategia via hill-climb sui parametri) e OGNI 10
|
|
EPOCHE blocca il 10% meno profittevole. Restano i 10 piu' profittevoli.
|
|
|
|
Punteggio = fitness su PNL + %win, con vincolo >=10 trade/mese (engine).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import random
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
from scripts.games.engine import load_anon, splits3, evaluate
|
|
|
|
OUT = Path("data/games")
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Spazio parametri per famiglia (min, max, tipo)
|
|
SPACE = {
|
|
"zscore": dict(lookback=(10, 100, "i"), entry_thr=(1.0, 3.5, "f"),
|
|
tp_atr=(0.5, 4.0, "f"), sl_atr=(1.0, 5.0, "f"),
|
|
max_bars=(6, 72, "i")),
|
|
"breakout": dict(lookback=(12, 120, "i"), entry_thr=(0.0, 0.0, "f"),
|
|
tp_atr=(0.5, 4.0, "f"), sl_atr=(1.0, 5.0, "f"),
|
|
max_bars=(6, 72, "i")),
|
|
"ma_cross": dict(lookback=(5, 50, "i"), slow_mult=(2.0, 6.0, "f"),
|
|
entry_thr=(0.0, 0.0, "f"), tp_atr=(0.5, 4.0, "f"),
|
|
sl_atr=(1.0, 5.0, "f"), max_bars=(6, 72, "i")),
|
|
"rsi": dict(lookback=(7, 30, "i"), entry_thr=(1.0, 4.0, "f"),
|
|
tp_atr=(0.5, 4.0, "f"), sl_atr=(1.0, 5.0, "f"),
|
|
max_bars=(6, 72, "i")),
|
|
"momentum": dict(lookback=(6, 72, "i"), entry_thr=(1.0, 6.0, "f"),
|
|
tp_atr=(0.5, 4.0, "f"), sl_atr=(1.0, 5.0, "f"),
|
|
max_bars=(6, 72, "i")),
|
|
"pairs": dict(lookback=(20, 120, "i"), entry_thr=(1.5, 3.0, "f"),
|
|
exit_thr=(0.2, 1.0, "f"), max_bars=(24, 120, "i")),
|
|
}
|
|
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":
|
|
return int(rng.randint(int(lo), int(hi)))
|
|
return round(rng.uniform(lo, hi), 3)
|
|
|
|
|
|
def random_spec(rng):
|
|
if not NO_LIVE and rng.random() < 0.25:
|
|
fam = "pairs"
|
|
else:
|
|
fam = rng.choice(SINGLE_FAMILIES)
|
|
params = {}
|
|
for k, (lo, hi, typ) in SPACE[fam].items():
|
|
params[k] = _rand_param(rng, lo, hi, typ)
|
|
spec = {"family": fam, "params": params, "tf": rng.choice(TIMEFRAMES)}
|
|
if fam == "pairs":
|
|
spec["series"] = "AB"
|
|
else:
|
|
spec["series"] = rng.choice(["A", "B"])
|
|
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
|
|
|
|
|
|
def mutate(spec, rng, strength=0.25):
|
|
"""Perturba la spec (hill-climb). Per lo piu' numerica; raramente
|
|
cambia direzione/serie. La famiglia resta fissa (identita' dell'agente)."""
|
|
s = json.loads(json.dumps(spec))
|
|
fam = s["family"]
|
|
# perturba 1-2 parametri numerici
|
|
keys = [k for k in SPACE[fam] if SPACE[fam][k][0] != SPACE[fam][k][1]]
|
|
for k in rng.sample(keys, k=min(len(keys), rng.randint(1, 2))):
|
|
lo, hi, typ = SPACE[fam][k]
|
|
cur = s["params"][k]
|
|
span = (hi - lo) * strength
|
|
nv = cur + rng.uniform(-span, span)
|
|
nv = max(lo, min(hi, nv))
|
|
s["params"][k] = int(round(nv)) if typ == "i" else round(nv, 3)
|
|
if fam != "pairs":
|
|
if rng.random() < 0.10:
|
|
s["params"]["direction"] = rng.choice(DIRECTIONS)
|
|
if rng.random() < 0.05:
|
|
s["series"] = rng.choice(["A", "B"])
|
|
# il timeframe resta l'identita' dell'agente (timing fisso) -> non muta
|
|
return s
|
|
|
|
|
|
def _normalize(spec):
|
|
"""Completa/ripulisce una spec proposta da un agente (robustezza)."""
|
|
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)
|
|
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)
|
|
out["tf"] = spec.get("tf") if spec.get("tf") in TIMEFRAMES else "1h"
|
|
if fam == "pairs":
|
|
out["series"] = "AB"
|
|
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")
|
|
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
|
|
|
|
|
|
class Agent:
|
|
def __init__(self, aid, spec, brief=""):
|
|
self.id = aid
|
|
self.spec = _normalize(spec)
|
|
self.brief = brief # cosa "dice" l'agente (ipotesi NL)
|
|
self.train_fit = -1e9 # criterio di hill-climb (l'agente ottimizza qui)
|
|
self.valid_fit = -1e9 # criterio dell'orchestratore (cull + rank)
|
|
self.metrics = {} # metriche TRAIN
|
|
self.vmetrics = {} # metriche VALID
|
|
self.alive = True
|
|
self.culled_epoch = None
|
|
|
|
@property
|
|
def tf(self):
|
|
return self.spec.get("tf", "1h")
|
|
|
|
def score(self, datasets, splits_map):
|
|
data = datasets[self.tf]
|
|
tr, va, _ = splits_map[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=7,
|
|
epochs=90, cull_every=10, cull_n=10, log=print,
|
|
out_name="tournament_result.json"):
|
|
rng = random.Random(seed)
|
|
# carica solo i timeframe effettivamente usati dagli agenti
|
|
used_tfs = sorted({_normalize(s).get("tf", "1h") for s in specs})
|
|
datasets = {tf: load_anon(tf) for tf in used_tfs}
|
|
splits_map = {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, splits_map)
|
|
|
|
alive = lambda: [a for a in agents if a.alive]
|
|
log(f"[epoch 0] {len(alive())} agenti | best VALID fit "
|
|
f"{max(a.valid_fit for a in agents):.1f}")
|
|
|
|
history = []
|
|
for ep in range(1, epochs + 1):
|
|
# elaborazione: l'agente affina sul TRAIN (cio' che vede); ricalcola VALID
|
|
for a in alive():
|
|
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 = cand
|
|
a.metrics, a.train_fit = m, m["fitness"]
|
|
a.vmetrics = evaluate(data, a.spec, va)
|
|
a.valid_fit = a.vmetrics["fitness"]
|
|
# cull ogni N epoche: l'ORCHESTRATORE blocca il 10% meno profittevole
|
|
# in VALIDATION (generalizzazione, non overfit sul train)
|
|
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)
|
|
# report finale: TEST = OOS puro mai toccato dall'ottimizzazione
|
|
results = []
|
|
for rank, a in enumerate(survivors, 1):
|
|
data = datasets[a.tf]
|
|
_, _, te = splits_map[a.tf]
|
|
test = evaluate(data, a.spec, te)
|
|
full = evaluate(data, a.spec, None)
|
|
results.append({
|
|
"rank": rank, "agent": a.id, "spec": a.spec, "brief": a.brief,
|
|
"tf": a.tf, "train": a.metrics, "valid": a.vmetrics,
|
|
"test": test, "full": full,
|
|
})
|
|
payload = {"n_agents": len(specs), "epochs": epochs,
|
|
"survivors": len(survivors), "results": results,
|
|
"history": history,
|
|
"reveal": {"A": "BTC", "B": "ETH", "tf": "1h"}}
|
|
(OUT / out_name).write_text(json.dumps(payload, indent=2))
|
|
return payload
|
|
|
|
|
|
def leaderboard(payload, top=10, log=print):
|
|
log("\n================ CLASSIFICA FINALE (top %d) ================" % top)
|
|
log("VALID = finestra su cui l'orchestratore giudica | TEST = OOS puro (mai ottimizzato)")
|
|
log(f"{'#':>2} {'ag':>4} {'tf':>3} {'famiglia':>9} {'ser':>3} {'dir':>9} "
|
|
f"{'TEpnl%':>8} {'TEwin':>5} {'TEtpm':>6} {'TEsh':>5} {'VApnl%':>8} {'VAwin':>5}")
|
|
for r in payload["results"][:top]:
|
|
sp = r["spec"]; te = r["test"]; va = r["valid"]
|
|
d = sp["params"].get("direction", "-")
|
|
log(f"{r['rank']:>2} {r['agent']:>4} {sp.get('tf','1h'):>3} {sp['family']:>9} "
|
|
f"{sp['series']:>3} {d:>9} {te['pnl_pct']:>8.0f} {te['win_rate']*100:>4.0f}% "
|
|
f"{te['tpm']:>6.1f} {te['sharpe']:>5.1f} {va['pnl_pct']:>8.0f} "
|
|
f"{va['win_rate']*100:>4.0f}%")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
# modalita' test: 100 agenti random
|
|
rng = random.Random(42)
|
|
specs = [random_spec(rng) for _ in range(100)]
|
|
payload = run_tournament(specs, seed=42)
|
|
leaderboard(payload)
|