diff --git a/scripts/strategies/DIP01_dip_buy.py b/scripts/strategies/DIP01_dip_buy.py new file mode 100644 index 0000000..9b60b50 --- /dev/null +++ b/scripts/strategies/DIP01_dip_buy.py @@ -0,0 +1,54 @@ +"""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) + 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: + 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 diff --git a/src/live/strategy_loader.py b/src/live/strategy_loader.py index dbdb522..fb75b27 100644 --- a/src/live/strategy_loader.py +++ b/src/live/strategy_loader.py @@ -17,6 +17,7 @@ _REGISTRY: dict[str, type[Strategy]] = {} # scripts/waste/: l'edge storico era un artefatto di look-ahead # (vedi scripts/analysis/oos_validation.py). MODULE_MAP = { + "DIP01_dip_buy": ("DIP01_dip_buy", "Dip01DipBuy"), "MR01_bollinger_fade": ("MR01_bollinger_fade", "BollingerFade"), "MR02_donchian_fade": ("MR02_donchian_fade", "DonchianFade"), "MR07_return_reversal": ("MR07_return_reversal", "ReturnReversal"), diff --git a/tests/portfolio/test_dip01.py b/tests/portfolio/test_dip01.py new file mode 100644 index 0000000..d49b39f --- /dev/null +++ b/tests/portfolio/test_dip01.py @@ -0,0 +1,13 @@ +import pandas as pd +from src.data.downloader import load_data +from scripts.strategies.DIP01_dip_buy import Dip01DipBuy + + +def test_dip01_generates_long_signals_with_exits(): + df = load_data("BTC", "1h").iloc[-5000:].reset_index(drop=True) + ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) + sigs = Dip01DipBuy().generate_signals(df, ts, asset="BTC", tf="1h") + assert len(sigs) > 0 + s = sigs[0] + assert s.direction == 1 + assert {"tp", "sl", "max_bars"} <= set(s.metadata)