Adriano Dal Pastro
d25d897fd1
Origine: gioco "Blind Traders" (100 agenti ciechi su BTC/ETH anonimizzati) -> vincitore = spread ETH/BTC reversion a 15m. Testato sul serio col gate PORT06: non duplicato (corr 1h vs 15m = 0.37), robusto (16/16 celle Sharpe>1), edge NON artefatto delle candele flat ETH 15m (filtrandole resta l'83% dello Sharpe). Percorso live costruito e validato: - pairs_research.pairs_sim_flat: engine generalizzato con exit LIVE-REALIZABLE (arma exit_ready, esce alla 1a barra pulita); regression-lock a pairs_sim. - PairsWorker: flat_skip + exit_ready + rilevamento flat da OHLC (1h byte-exact). - runner: fetch diretto dei timeframe sub-orari + override position_size per-sleeve. - validate_worker_pairs: replay worker == backtest a 15m (8452 vs 8453 trade). - _defs/build_everything: sleeve PR_ETHBTC_15M (mezza size, pos 0.10) -> PORT06 FULL 6.43->7.20, OOS 8.58->9.66, DD giu'. Rischio bilanciato col 1h. - smoke live: Cerbero serve candele 15m fresche; worker ticca. Diari docs/diary/2026-06-09-*. Caveat slippage: mezza size = blend-tilt prudente. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
324 lines
11 KiB
Python
324 lines
11 KiB
Python
"""
|
|
Game engine — "Blind Traders" tournament.
|
|
|
|
100 agenti ricevono due serie anonime (A, B) — in realta' BTC e ETH 1h — e
|
|
propongono strategie senza sapere cosa sono. L'orchestratore (questo motore)
|
|
valuta ogni strategia con un backtest deterministico, causale e fee-aware, e
|
|
assegna un punteggio su %win + PNL con vincolo >=10 trade/mese.
|
|
|
|
Tutto causale (nessun look-ahead): i segnali alla barra i usano solo dati
|
|
fino a close[i]; l'ingresso e' a close[i], le uscite TP/SL/max_bars intrabar
|
|
dalle barre successive.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from src.data.downloader import load_data
|
|
|
|
FEE_RT = 0.001 # 0.10% round-trip (taker Deribit, baseline progetto)
|
|
TF_BPM = {"5m": 12 * 24 * 30, "15m": 4 * 24 * 30, "1h": 24 * 30} # barre/mese per tf
|
|
MIN_TRADES_PER_MONTH = 10.0
|
|
|
|
# Slippage per LATO (oltre alle fee). 0 = come prima. Single-leg paga 2 lati
|
|
# (ingresso+uscita), i pairs ne pagano 4 (2 gambe x 2 lati).
|
|
_SLIP = 0.0
|
|
|
|
|
|
def set_slippage(slip_per_side: float):
|
|
global _SLIP
|
|
_SLIP = float(slip_per_side)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Dati anonimizzati
|
|
# --------------------------------------------------------------------------
|
|
def load_anon(tf: str = "1h"):
|
|
"""Carica BTC->A, ETH->B allineati sull'intersezione temporale.
|
|
|
|
Ritorna un dict con array OHLC per A e B + datetime. I nomi reali NON
|
|
compaiono: gli agenti vedono solo 'A' e 'B'.
|
|
"""
|
|
btc = load_data("BTC", tf).copy()
|
|
eth = load_data("ETH", tf).copy()
|
|
for d in (btc, eth):
|
|
d["dt"] = pd.to_datetime(d["datetime"])
|
|
btc = btc.set_index("dt")
|
|
eth = eth.set_index("dt")
|
|
idx = btc.index.intersection(eth.index)
|
|
btc = btc.loc[idx].sort_index()
|
|
eth = eth.loc[idx].sort_index()
|
|
out = {"dt": idx.to_numpy()}
|
|
for name, d in (("A", btc), ("B", eth)):
|
|
out[name] = {
|
|
"open": d["open"].to_numpy(float),
|
|
"high": d["high"].to_numpy(float),
|
|
"low": d["low"].to_numpy(float),
|
|
"close": d["close"].to_numpy(float),
|
|
"volume": d["volume"].to_numpy(float),
|
|
}
|
|
out["n"] = len(idx)
|
|
out["tf"] = tf
|
|
out["bpm"] = TF_BPM[tf]
|
|
return out
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Indicatori causali (vettorizzati)
|
|
# --------------------------------------------------------------------------
|
|
def _roll_mean(x, w):
|
|
return pd.Series(x).rolling(w).mean().to_numpy()
|
|
|
|
|
|
def _roll_std(x, w):
|
|
return pd.Series(x).rolling(w).std(ddof=0).to_numpy()
|
|
|
|
|
|
def _ema(x, w):
|
|
return pd.Series(x).ewm(span=w, adjust=False).mean().to_numpy()
|
|
|
|
|
|
def _atr(high, low, close, w=14):
|
|
pc = np.roll(close, 1)
|
|
pc[0] = close[0]
|
|
tr = np.maximum(high - low, np.maximum(np.abs(high - pc), np.abs(low - pc)))
|
|
return pd.Series(tr).rolling(w).mean().to_numpy()
|
|
|
|
|
|
def _rsi(close, w=14):
|
|
d = np.diff(close, prepend=close[0])
|
|
up = np.where(d > 0, d, 0.0)
|
|
dn = np.where(d < 0, -d, 0.0)
|
|
ru = pd.Series(up).ewm(alpha=1 / w, adjust=False).mean().to_numpy()
|
|
rd = pd.Series(dn).ewm(alpha=1 / w, adjust=False).mean().to_numpy()
|
|
rs = ru / (rd + 1e-12)
|
|
return 100 - 100 / (1 + rs)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Famiglie di segnale -> array di posizione desiderata {-1,0,+1} alla barra i
|
|
# (causale: usa solo dati fino a close[i]). +1 = long, -1 = short.
|
|
# --------------------------------------------------------------------------
|
|
def _signal_single(o, family, p):
|
|
"""Segnale per una singola serie. Ritorna (pos_target, atr)."""
|
|
close = o["close"]
|
|
high, low = o["high"], o["low"]
|
|
n = len(close)
|
|
atr = _atr(high, low, close, 14)
|
|
pos = np.zeros(n)
|
|
lb = max(2, int(p["lookback"]))
|
|
thr = float(p["entry_thr"])
|
|
sign = 1 if p.get("direction", "reversion") == "trend" else -1
|
|
|
|
if family == "zscore":
|
|
ma = _roll_mean(close, lb)
|
|
sd = _roll_std(close, lb)
|
|
z = (close - ma) / (sd + 1e-12)
|
|
pos = np.where(z > thr, sign * -1.0, np.where(z < -thr, sign * 1.0, 0.0))
|
|
elif family == "breakout":
|
|
hh = pd.Series(high).rolling(lb).max().shift(1).to_numpy()
|
|
ll = pd.Series(low).rolling(lb).min().shift(1).to_numpy()
|
|
up = close > hh
|
|
dn = close < ll
|
|
# trend: break-up=long ; reversion: break-up=short
|
|
pos = np.where(up, sign * 1.0, np.where(dn, sign * -1.0, 0.0))
|
|
elif family == "ma_cross":
|
|
fast = _ema(close, lb)
|
|
slow = _ema(close, max(lb + 2, int(lb * p.get("slow_mult", 3))))
|
|
pos = np.where(fast > slow, sign * 1.0, sign * -1.0)
|
|
elif family == "rsi":
|
|
r = _rsi(close, lb)
|
|
hi = 50 + thr * 10
|
|
lo = 50 - thr * 10
|
|
pos = np.where(r > hi, sign * -1.0, np.where(r < lo, sign * 1.0, 0.0))
|
|
elif family == "momentum":
|
|
ret = close / np.roll(close, lb) - 1
|
|
ret[:lb] = 0
|
|
pos = np.where(ret > thr / 100, sign * 1.0,
|
|
np.where(ret < -thr / 100, sign * -1.0, 0.0))
|
|
else:
|
|
raise ValueError(f"unknown family {family}")
|
|
pos = np.nan_to_num(pos)
|
|
return pos, atr
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Backtest single-series (long/short con TP/SL/max_bars intrabar)
|
|
# --------------------------------------------------------------------------
|
|
def _backtest_single(o, pos, atr, p, fee=FEE_RT):
|
|
close, high, low = o["close"], o["high"], o["low"]
|
|
n = len(close)
|
|
tp_atr = float(p.get("tp_atr", 2.0))
|
|
sl_atr = float(p.get("sl_atr", 2.0))
|
|
max_bars = int(p.get("max_bars", 24))
|
|
rets = [] # net return per trade
|
|
# warmup
|
|
start = max(int(p["lookback"]) + 15, 20)
|
|
# indici candidati: solo barre con segnale != 0 (salta le barre flat)
|
|
cand = np.flatnonzero(pos[start:n - 1]) + start
|
|
ci = 0
|
|
nc = len(cand)
|
|
while ci < nc:
|
|
i = int(cand[ci])
|
|
d = pos[i]
|
|
if d == 0 or np.isnan(atr[i]) or atr[i] <= 0:
|
|
ci += 1
|
|
continue
|
|
entry = close[i]
|
|
a = atr[i]
|
|
if d > 0:
|
|
tp = entry + tp_atr * a
|
|
sl = entry - sl_atr * a
|
|
else:
|
|
tp = entry - tp_atr * a
|
|
sl = entry + sl_atr * a
|
|
exit_px = None
|
|
j = i + 1
|
|
end = min(n - 1, i + max_bars)
|
|
while j <= end:
|
|
hi, lo = high[j], low[j]
|
|
if d > 0:
|
|
if lo <= sl: # SL prioritario
|
|
exit_px = sl
|
|
break
|
|
if hi >= tp:
|
|
exit_px = tp
|
|
break
|
|
else:
|
|
if hi >= sl:
|
|
exit_px = sl
|
|
break
|
|
if lo <= tp:
|
|
exit_px = tp
|
|
break
|
|
j += 1
|
|
if exit_px is None:
|
|
exit_px = close[end]
|
|
j = end
|
|
gross = d * (exit_px - entry) / entry
|
|
net = gross - fee - 2 * _SLIP # 2 lati di slippage
|
|
rets.append(net)
|
|
# salta al primo ingresso candidato OLTRE l'uscita (no overlap)
|
|
ci = int(np.searchsorted(cand, j + 1, side="left"))
|
|
return np.array(rets)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Backtest cross-series (pairs market-neutral sullo z del log-ratio)
|
|
# --------------------------------------------------------------------------
|
|
def _backtest_pairs(A, B, p, fee=FEE_RT):
|
|
a, b = A["close"], B["close"]
|
|
n = len(a)
|
|
lb = max(5, int(p["lookback"]))
|
|
z_in = float(p["entry_thr"])
|
|
z_exit = float(p.get("exit_thr", 0.5))
|
|
max_bars = int(p.get("max_bars", 72))
|
|
lr = np.log(a / b)
|
|
ma = _roll_mean(lr, lb)
|
|
sd = _roll_std(lr, lb)
|
|
z = (lr - ma) / (sd + 1e-12)
|
|
rets = []
|
|
start = max(lb + 5, 20)
|
|
zabs = np.abs(z)
|
|
zabs[:start] = 0.0
|
|
zabs[np.isnan(zabs)] = 0.0
|
|
cand = np.flatnonzero(zabs[:n - 1] > z_in)
|
|
ci = 0
|
|
nc = len(cand)
|
|
while ci < nc:
|
|
i = int(cand[ci])
|
|
d = -1 if z[i] > z_in else 1 # spread alto -> short A/long B ; basso -> long A/short B
|
|
ea, eb = a[i], b[i]
|
|
j = i + 1
|
|
end = min(n - 1, i + max_bars)
|
|
while j <= end:
|
|
if abs(z[j]) <= z_exit:
|
|
break
|
|
j += 1
|
|
j = min(j, end)
|
|
# PnL = gamba A (dir d) + gamba B (dir -d), fee su 2 gambe
|
|
ra = d * (a[j] - ea) / ea
|
|
rb = -d * (b[j] - eb) / eb
|
|
net = ra + rb - 2 * fee - 4 * _SLIP # 2 gambe x 2 lati di slippage
|
|
rets.append(net)
|
|
ci = int(np.searchsorted(cand, j + 1, side="left"))
|
|
return np.array(rets)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Valutazione + scoring
|
|
# --------------------------------------------------------------------------
|
|
def evaluate(data, spec, sl=None, fee=FEE_RT):
|
|
"""Valuta una spec di strategia su uno slice [start,end) (sl=slice di indici).
|
|
|
|
spec = {family, series, params{...}}. Ritorna dict metriche.
|
|
"""
|
|
family = spec["family"]
|
|
series = spec.get("series", "A")
|
|
p = spec["params"]
|
|
|
|
def _slice(o):
|
|
if sl is None:
|
|
return o
|
|
s, e = sl
|
|
return {k: v[s:e] for k, v in o.items()}
|
|
|
|
if family == "pairs":
|
|
A = _slice(data["A"])
|
|
B = _slice(data["B"])
|
|
rets = _backtest_pairs(A, B, p, fee)
|
|
nbars = len(A["close"])
|
|
else:
|
|
o = _slice(data[series])
|
|
pos, atr = _signal_single(o, family, p)
|
|
rets = _backtest_single(o, pos, atr, p, fee)
|
|
nbars = len(o["close"])
|
|
|
|
n_tr = len(rets)
|
|
months = nbars / data.get("bpm", TF_BPM["1h"])
|
|
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_rate = float(np.mean(rets > 0))
|
|
pnl = float(np.sum(rets)) * 100 # PnL additivo (notional fisso), %
|
|
equity = float(np.prod(1 + rets) - 1) * 100 # equity compounding, %
|
|
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 domina, win% come spinta secondaria; squalifica se pochi trade
|
|
fitness = pnl + 50.0 * win_rate
|
|
if not qualified:
|
|
fitness = -1e6 + pnl # ordinati ma fuori gioco
|
|
return dict(n_trades=n_tr, win_rate=win_rate, pnl_pct=pnl, equity_pct=equity,
|
|
tpm=tpm, sharpe=sharpe, avg_ret=avg, qualified=qualified,
|
|
fitness=fitness)
|
|
|
|
|
|
# Split a 3: TRAIN (hill-climb) / VALID (cull+rank dell'orchestratore) / TEST (OOS puro)
|
|
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)
|
|
|
|
|
|
# compat: split a 2 (train/oos)
|
|
def splits(data, train_frac=0.70):
|
|
n = data["n"]
|
|
cut = int(n * train_frac)
|
|
return (0, cut), (cut, n)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
data = load_anon("1h")
|
|
print("loaded", data["n"], "bars,", data["dt"][0], "->", data["dt"][-1])
|
|
tr, oos = splits(data)
|
|
demo = {"family": "zscore", "series": "B",
|
|
"params": {"lookback": 20, "entry_thr": 2.0, "direction": "reversion",
|
|
"tp_atr": 1.5, "sl_atr": 2.0, "max_bars": 24}}
|
|
print("TRAIN", evaluate(data, demo, tr))
|
|
print("OOS ", evaluate(data, demo, oos))
|