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>
86 lines
3.3 KiB
Python
86 lines
3.3 KiB
Python
"""EXIT-03 — trail_pct: trailing percentuale dall'high-water-mark favorevole (sui CLOSE).
|
|
|
|
Idea: invece di un trail ATR (EXIT-01/02), lo stop segue l'estremo favorevole
|
|
RAGGIUNTO dai CLOSE a una distanza percentuale fissa p. Lo stop puo' solo
|
|
stringersi (mai allentarsi sotto sl0 lato rischio).
|
|
|
|
Long : hwm = max(close[i..j-1]); stop(j) = max(sl0, hwm*(1-p)) (sale, mai scende)
|
|
Short: lwm = min(close[i..j-1]); stop(j) = min(sl0, lwm*(1+p)) (scende, mai sale)
|
|
|
|
Due varianti:
|
|
keep_tp=True -> il TP fisso tp0 dall'entrata RESTA attivo (esci al primo dei due),
|
|
horizon = max_bars (invariato). Il trail morde solo se il TP non
|
|
viene raggiunto e il prezzo ha prima corso a favore e poi ritracciato.
|
|
keep_tp=False -> TP RIMOSSO (cavalcata pura), horizon = 2*max_bars (cap HARD_CAP).
|
|
Si esce solo sul trailing-stop o a orizzonte.
|
|
|
|
ANTI-LOOK-AHEAD: levels(j) usa SOLO dati <= j-1:
|
|
- hwm/lwm sullo slice [i..j-1] dei CLOSE (mantenuto incrementalmente col bar j-1);
|
|
- p e' una costante -> nessuna dipendenza da indicatori del bar j.
|
|
L'hwm sui close (non sugli high) e' deliberato: il close e' il dato su cui il
|
|
worker live decide al poll del tick; un hwm sugli high anticiperebbe un livello
|
|
che il close non ha ancora confermato.
|
|
|
|
GRID: p in {0.005, 0.01, 0.02} x keep_tp in {True, False} (6 celle).
|
|
"""
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from exit_lab import ExitPolicy, evaluate, HARD_CAP # noqa: E402
|
|
|
|
|
|
class TrailPct(ExitPolicy):
|
|
name = "trail_pct"
|
|
|
|
def __init__(self, ctx, i, d, entry, tp0, sl0, mb,
|
|
p=0.01, keep_tp=True, **params):
|
|
super().__init__(ctx, i, d, entry, tp0, sl0, mb, **params)
|
|
self.p = float(p)
|
|
self.keep_tp = bool(keep_tp)
|
|
if not self.keep_tp:
|
|
self.horizon = min(2 * mb, HARD_CAP)
|
|
self.close = ctx["close"]
|
|
# high/low-water-mark sui CLOSE, solo barre <= j-1; init = close d'entrata i
|
|
self.hwm = self.close[i]
|
|
self.lwm = self.close[i]
|
|
self.last_seen = i # ultimo indice gia' incorporato
|
|
# stop trailing monotono: parte da sl0 e puo' solo stringersi
|
|
self.cur_stop = sl0
|
|
|
|
def _update_wm(self, upto: int) -> None:
|
|
"""Incorpora le barre (last_seen, upto] nell'estremo sui close. upto = j-1
|
|
-> NON tocca il bar j (anti-look-ahead)."""
|
|
while self.last_seen < upto:
|
|
self.last_seen += 1
|
|
cv = self.close[self.last_seen]
|
|
if cv > self.hwm:
|
|
self.hwm = cv
|
|
if cv < self.lwm:
|
|
self.lwm = cv
|
|
|
|
def levels(self, j: int):
|
|
self._update_wm(j - 1) # solo dati <= j-1
|
|
if self.d == 1:
|
|
cand = self.hwm * (1.0 - self.p)
|
|
if cand > self.cur_stop: # lo stop long puo' solo SALIRE
|
|
self.cur_stop = cand
|
|
else:
|
|
cand = self.lwm * (1.0 + self.p)
|
|
if cand < self.cur_stop: # lo stop short puo' solo SCENDERE
|
|
self.cur_stop = cand
|
|
tp = self.tp0 if self.keep_tp else None
|
|
return tp, self.cur_stop, 1.0
|
|
|
|
|
|
GRID = [
|
|
{"p": p, "keep_tp": kt}
|
|
for p in (0.005, 0.01, 0.02)
|
|
for kt in (True, False)
|
|
]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
evaluate(TrailPct, GRID)
|