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>
59 lines
2.4 KiB
Python
59 lines
2.4 KiB
Python
"""DIP01 — Dip-buy mean-reversion single-asset (z-score sotto-banda). Honest family.
|
|
|
|
Replica live della logica validata in scripts/analysis/honest_improve2.dip_market_gated
|
|
(con market_n=0, come lo sleeve DIP01_BTC del portafoglio): compra quando lo z-score del
|
|
prezzo rispetto a SMA(n) incrocia sotto -z_in; esce a TP=SMA, SL=close-sl_atr*ATR, o max_bars.
|
|
"""
|
|
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.strategies.base import Strategy, Signal # noqa: E402
|
|
|
|
|
|
def _atr(df, n=14):
|
|
h, l, c = df["high"].values, df["low"].values, df["close"].values
|
|
pc = np.roll(c, 1); pc[0] = c[0]
|
|
tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
|
|
return pd.Series(tr).rolling(n).mean().values
|
|
|
|
|
|
class Dip01DipBuy(Strategy):
|
|
name = "DIP01_dip_buy"
|
|
description = "Dip-buy mean-reversion single-asset (z-score), exit TP=SMA/SL=ATR/max_bars"
|
|
default_assets = ["BTC"]
|
|
default_timeframes = ["1h"]
|
|
fee_rt = 0.001
|
|
leverage = 3.0
|
|
position_size = 0.15
|
|
|
|
def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex,
|
|
n: int = 50, z_in: float = 2.5, sl_atr: float = 2.5,
|
|
max_bars: int = 24, **params) -> list[Signal]:
|
|
c = df["close"].values
|
|
ma = pd.Series(c).rolling(n).mean().values
|
|
sd = pd.Series(c).rolling(n).std().values
|
|
a = _atr(df, 14)
|
|
z = (c - ma) / np.where(sd == 0, np.nan, sd)
|
|
# Edge minimo: salta i dip il cui TP (la media) è entro il costo round-trip. 0 = off.
|
|
min_tp_frac = params.get("min_tp_frac", 0.0)
|
|
out: list[Signal] = []
|
|
for i in range(n + 14, len(c)):
|
|
if np.isnan(z[i]) or np.isnan(a[i]) or np.isnan(ma[i]):
|
|
continue
|
|
if z[i] <= -z_in and z[i - 1] > -z_in:
|
|
if min_tp_frac > 0 and abs(ma[i] - c[i]) / c[i] <= min_tp_frac:
|
|
continue # TP entro le fee -> non eseguibile in utile
|
|
out.append(Signal(idx=i, direction=1, entry_price=float(c[i]),
|
|
metadata={"tp": float(ma[i]),
|
|
"sl": float(c[i] - sl_atr * a[i]),
|
|
"max_bars": int(max_bars)}))
|
|
return out
|