feat(live): DIP01 dip-buy come Strategy single-asset (worker via StrategyWorker)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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

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"),

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)