feat: paper trading live su Deribit testnet — squeeze+ML ibrida

Sistema completo: client Cerbero MCP, signal engine (squeeze + GBM),
paper trader con gestione posizioni, stop loss, log JSONL.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

scripts/paper_status.py

@@ -0,0 +1,79 @@
+"""Mostra lo stato del paper trader."""
+from __future__ import annotations
+import sys
+sys.path.insert(0, ".")
+
+import json
+from pathlib import Path
+from src.live.cerbero_client import CerberoClient
+
+LOG_DIR = Path("data/paper_trades")
+
+print("=" * 50)
+print("  PAPER TRADER STATUS")
+print("=" * 50)
+
+# Status file
+status_path = LOG_DIR / "status.json"
+if status_path.exists():
+    with open(status_path) as f:
+        status = json.load(f)
+    print(f"\n  In posizione: {status['in_position']}")
+    if status["in_position"]:
+        print(f"  Direzione:    {status['direction']}")
+        print(f"  Entry price:  {status['entry_price']}")
+        print(f"  Entry time:   {status['entry_time']}")
+        print(f"  Barre tenute: {status['bars_held']}")
+    print(f"  Ultimo update: {status['last_update']}")
+else:
+    print("\n  Nessun file di stato trovato.")
+
+# Account
+print("\n--- ACCOUNT DERIBIT TESTNET ---")
+c = CerberoClient()
+try:
+    acc = c.get_account_summary()
+    print(f"  Equity:  {acc['equity']:.2f}")
+    print(f"  Balance: {acc['balance']:.2f}")
+    print(f"  PnL:     {acc['total_pnl']:.2f}")
+except Exception as e:
+    print(f"  Errore: {e}")
+
+# Posizioni
+try:
+    pos = c.get_positions()
+    print(f"\n  Posizioni aperte: {len(pos)}")
+    for p in pos:
+        print(f"    {p}")
+except Exception as e:
+    print(f"  Errore: {e}")
+
+# Ultimi log
+print("\n--- ULTIMI LOG ---")
+log_files = sorted(LOG_DIR.glob("trades_*.jsonl"))
+if log_files:
+    with open(log_files[-1]) as f:
+        lines = f.readlines()
+    for line in lines[-10:]:
+        entry = json.loads(line)
+        print(f"  [{entry['timestamp'][:19]}] {entry['event']}")
+else:
+    print("  Nessun log trovato.")
+
+# Statistiche trade
+all_trades = []
+for lf in log_files:
+    with open(lf) as f:
+        for line in f:
+            entry = json.loads(line)
+            if entry["event"] == "CLOSED":
+                all_trades.append(entry)
+
+if all_trades:
+    wins = sum(1 for t in all_trades if t.get("pnl_pct", 0) > 0)
+    total = len(all_trades)
+    total_pnl = sum(t.get("pnl_pct", 0) for t in all_trades)
+    print(f"\n--- STATISTICHE ---")
+    print(f"  Trade chiusi: {total}")
+    print(f"  Win rate:     {wins/total*100:.0f}%")
+    print(f"  PnL totale:   {total_pnl:.2f}%")

src/live/cerbero_client.py

@@ -0,0 +1,93 @@
+"""Client HTTP per Cerbero MCP — Deribit testnet."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+import requests
+
+BASE_URL = "https://cerbero-mcp.tielogic.xyz"
+TOKEN = "_hm0FkyC67P9OXJTy7R9SE2lfhGz_Wa6i89KqH_uXrk"
+BOT_TAG = "pythagoras-paper"
+TIMEOUT = 15
+
+
+@dataclass
+class CerberoClient:
+    base_url: str = BASE_URL
+    token: str = TOKEN
+    bot_tag: str = BOT_TAG
+
+    def _headers(self) -> dict[str, str]:
+        return {
+            "Authorization": f"Bearer {self.token}",
+            "X-Bot-Tag": self.bot_tag,
+            "Content-Type": "application/json",
+        }
+
+    def _post(self, path: str, payload: dict | None = None) -> dict:
+        resp = requests.post(
+            f"{self.base_url}{path}",
+            headers=self._headers(),
+            json=payload or {},
+            timeout=TIMEOUT,
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+    # --- Market data ---
+
+    def get_ticker(self, instrument: str = "ETH-PERPETUAL") -> dict:
+        return self._post("/mcp-deribit/tools/get_ticker", {"instrument": instrument})
+
+    def get_historical(self, instrument: str, start_date: str, end_date: str, resolution: str = "15") -> list[dict]:
+        data = self._post("/mcp-deribit/tools/get_historical", {
+            "instrument": instrument,
+            "start_date": start_date,
+            "end_date": end_date,
+            "resolution": resolution,
+        })
+        return data.get("candles", [])
+
+    # --- Account ---
+
+    def get_account_summary(self) -> dict:
+        return self._post("/mcp-deribit/tools/get_account_summary")
+
+    def get_positions(self) -> list[dict]:
+        return self._post("/mcp-deribit/tools/get_positions")
+
+    # --- Trading ---
+
+    def place_order(
+        self,
+        instrument: str,
+        side: str,
+        amount: float,
+        order_type: str = "market",
+        price: float | None = None,
+        leverage: int | None = 3,
+        label: str | None = None,
+    ) -> dict:
+        payload: dict[str, Any] = {
+            "instrument_name": instrument,
+            "side": side,
+            "amount": amount,
+            "type": order_type,
+        }
+        if price is not None:
+            payload["price"] = price
+        if leverage is not None:
+            payload["leverage"] = leverage
+        if label:
+            payload["label"] = label
+        return self._post("/mcp-deribit/tools/place_order", payload)
+
+    def close_position(self, instrument: str) -> dict:
+        return self._post("/mcp-deribit/tools/close_position", {"instrument_name": instrument})
+
+    def set_stop_loss(self, order_id: str, stop_price: float) -> dict:
+        return self._post("/mcp-deribit/tools/set_stop_loss", {"order_id": order_id, "stop_price": stop_price})
+
+    def set_take_profit(self, order_id: str, tp_price: float) -> dict:
+        return self._post("/mcp-deribit/tools/set_take_profit", {"order_id": order_id, "tp_price": tp_price})

src/live/paper_trader.py

@@ -0,0 +1,250 @@
+"""Paper trader: loop principale che monitora, segnala e opera su Deribit testnet."""
+from __future__ import annotations
+
+import json
+import time
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pandas as pd
+
+from src.live.cerbero_client import CerberoClient
+from src.live.signal_engine import SignalEngine
+
+LOG_DIR = Path(__file__).resolve().parents[2] / "data" / "paper_trades"
+INSTRUMENT = "ETH-PERPETUAL"
+RESOLUTION = "15"
+LEVERAGE = 3
+POSITION_PCT = 0.15
+HOLD_BARS = 3
+POLL_SECONDS = 60
+LOOKBACK_DAYS = 60
+TRAIN_LOOKBACK_DAYS = 365
+
+
+class PaperTrader:
+    def __init__(self):
+        self.client = CerberoClient()
+        self.engine = SignalEngine(bb_w=14, sq_thr=0.8, ml_thr=0.70)
+
+        self.in_position = False
+        self.position_entry_time: datetime | None = None
+        self.position_direction: str | None = None
+        self.position_entry_price: float = 0
+        self.bars_held = 0
+        self.last_bar_ts: int = 0
+
+        LOG_DIR.mkdir(parents=True, exist_ok=True)
+        self.log_path = LOG_DIR / f"trades_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jsonl"
+        self.status_path = LOG_DIR / "status.json"
+
+    def log(self, event: str, data: dict | None = None):
+        entry = {
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "event": event,
+            **(data or {}),
+        }
+        with open(self.log_path, "a") as f:
+            f.write(json.dumps(entry) + "\n")
+        print(f"  [{entry['timestamp'][:19]}] {event}: {json.dumps(data or {})}")
+
+    def save_status(self):
+        status = {
+            "in_position": self.in_position,
+            "direction": self.position_direction,
+            "entry_price": self.position_entry_price,
+            "entry_time": self.position_entry_time.isoformat() if self.position_entry_time else None,
+            "bars_held": self.bars_held,
+            "last_update": datetime.now(timezone.utc).isoformat(),
+        }
+        with open(self.status_path, "w") as f:
+            json.dump(status, f, indent=2)
+
+    def fetch_candles(self, days: int = LOOKBACK_DAYS) -> pd.DataFrame:
+        end = datetime.now(timezone.utc)
+        start = end - timedelta(days=days)
+        candles = self.client.get_historical(
+            INSTRUMENT,
+            start.strftime("%Y-%m-%d"),
+            end.strftime("%Y-%m-%d"),
+            RESOLUTION,
+        )
+        if not candles:
+            return pd.DataFrame()
+        df = pd.DataFrame(candles)
+        df["timestamp"] = df["timestamp"].astype("int64")
+        df = df.sort_values("timestamp").reset_index(drop=True)
+        return df
+
+    def train_model(self):
+        self.log("TRAINING", {"lookback_days": TRAIN_LOOKBACK_DAYS})
+        df = self.fetch_candles(TRAIN_LOOKBACK_DAYS)
+        if df.empty:
+            self.log("TRAINING_FAILED", {"reason": "no data"})
+            return False
+        result = self.engine.train(df, lookahead=HOLD_BARS)
+        self.log("TRAINING_DONE", result)
+        return "error" not in result
+
+    def open_position(self, direction: str, signal: dict):
+        ticker = self.client.get_ticker(INSTRUMENT)
+        price = ticker["last_price"]
+        account = self.client.get_account_summary()
+        equity = account["equity"]
+
+        notional = equity * POSITION_PCT
+        amount = round(notional / price, 1)
+        amount = max(amount, 1.0)
+
+        side = "buy" if direction == "buy" else "sell"
+
+        self.log("OPENING", {
+            "side": side,
+            "amount": amount,
+            "price": price,
+            "equity": equity,
+            "signal": signal,
+        })
+
+        try:
+            result = self.client.place_order(
+                instrument=INSTRUMENT,
+                side=side,
+                amount=amount,
+                order_type="market",
+                leverage=LEVERAGE,
+                label="pythagoras-squeeze",
+            )
+            self.in_position = True
+            self.position_direction = side
+            self.position_entry_price = price
+            self.position_entry_time = datetime.now(timezone.utc)
+            self.bars_held = 0
+            self.log("OPENED", {"order_result": result})
+        except Exception as e:
+            self.log("OPEN_FAILED", {"error": str(e)})
+
+    def close_current_position(self, reason: str):
+        if not self.in_position:
+            return
+
+        ticker = self.client.get_ticker(INSTRUMENT)
+        exit_price = ticker["last_price"]
+
+        if self.position_direction == "buy":
+            pnl_pct = (exit_price - self.position_entry_price) / self.position_entry_price * 100
+        else:
+            pnl_pct = (self.position_entry_price - exit_price) / self.position_entry_price * 100
+
+        self.log("CLOSING", {
+            "reason": reason,
+            "entry_price": self.position_entry_price,
+            "exit_price": exit_price,
+            "pnl_pct": round(pnl_pct, 3),
+            "bars_held": self.bars_held,
+        })
+
+        try:
+            result = self.client.close_position(INSTRUMENT)
+            self.log("CLOSED", {"result": result, "pnl_pct": round(pnl_pct, 3)})
+        except Exception as e:
+            self.log("CLOSE_FAILED", {"error": str(e)})
+
+        self.in_position = False
+        self.position_direction = None
+        self.position_entry_price = 0
+        self.position_entry_time = None
+        self.bars_held = 0
+
+    def check_position_exit(self, df: pd.DataFrame):
+        if not self.in_position:
+            return
+
+        current_ts = df["timestamp"].iloc[-1]
+        if current_ts > self.last_bar_ts:
+            self.bars_held += 1
+            self.last_bar_ts = current_ts
+
+        if self.bars_held >= HOLD_BARS:
+            self.close_current_position("hold_limit")
+            return
+
+        price = df["close"].iloc[-1]
+        if self.position_direction == "buy":
+            pnl_pct = (price - self.position_entry_price) / self.position_entry_price
+        else:
+            pnl_pct = (self.position_entry_price - price) / self.position_entry_price
+
+        if pnl_pct <= -0.02:
+            self.close_current_position("stop_loss_2pct")
+
+    def run_once(self) -> str:
+        """Esegui un singolo ciclo. Ritorna lo stato."""
+        df = self.fetch_candles(LOOKBACK_DAYS)
+        if df.empty:
+            return "no_data"
+
+        if self.in_position:
+            self.check_position_exit(df)
+            self.save_status()
+            if self.in_position:
+                return f"in_position_{self.position_direction}_bar{self.bars_held}"
+            return "position_closed"
+
+        signal = self.engine.check_signal(df)
+        if signal:
+            self.log("SIGNAL", signal)
+            self.open_position(signal["direction"], signal)
+            self.save_status()
+            return f"signal_{signal['direction']}"
+
+        self.save_status()
+        return "watching"
+
+    def run(self, retrain_hours: int = 24):
+        """Loop principale."""
+        print("=" * 60)
+        print(f"  PAPER TRADER — {INSTRUMENT} {RESOLUTION}m")
+        print(f"  Leva: {LEVERAGE}x, Position: {POSITION_PCT*100:.0f}%, Hold: {HOLD_BARS} barre")
+        print(f"  Poll: ogni {POLL_SECONDS}s")
+        print(f"  Log: {self.log_path}")
+        print("=" * 60)
+
+        account = self.client.get_account_summary()
+        self.log("STARTUP", {
+            "equity": account["equity"],
+            "testnet": account.get("testnet", True),
+        })
+
+        if not self.train_model():
+            print("Training fallito. Uscita.")
+            return
+
+        last_train = datetime.now(timezone.utc)
+
+        while True:
+            try:
+                now = datetime.now(timezone.utc)
+                if (now - last_train).total_seconds() > retrain_hours * 3600:
+                    self.train_model()
+                    last_train = now
+
+                status = self.run_once()
+                if status != "watching":
+                    print(f"  → {status}")
+
+            except KeyboardInterrupt:
+                self.log("SHUTDOWN", {"reason": "keyboard"})
+                if self.in_position:
+                    self.close_current_position("shutdown")
+                break
+            except Exception as e:
+                self.log("ERROR", {"error": str(e)})
+                print(f"  ERRORE: {e}")
+
+            time.sleep(POLL_SECONDS)
+
+
+if __name__ == "__main__":
+    trader = PaperTrader()
+    trader.run()

src/live/signal_engine.py

@@ -0,0 +1,232 @@
+"""Motore segnali: squeeze detection + ML confirmation su dati live."""
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+from sklearn.ensemble import GradientBoostingClassifier
+from sklearn.preprocessing import StandardScaler
+
+
+def keltner_ratio(close: np.ndarray, high: np.ndarray, low: np.ndarray, window: int = 14) -> np.ndarray:
+    n = len(close)
+    result = np.full(n, np.nan)
+    for i in range(window, n):
+        wc = close[i - window : i]
+        wh = high[i - window : i]
+        wl = low[i - window : i]
+        ma = np.mean(wc)
+        bb_std = np.std(wc)
+        tr = np.maximum(wh - wl, np.maximum(np.abs(wh - np.roll(wc, 1)), np.abs(wl - np.roll(wc, 1))))
+        atr = np.mean(tr[1:])
+        kc_r = (ma + 1.5 * atr) - (ma - 1.5 * atr)
+        bb_r = (ma + 2 * bb_std) - (ma - 2 * bb_std)
+        if kc_r > 0:
+            result[i] = bb_r / kc_r
+    return result
+
+
+def build_features(df: pd.DataFrame, i: int, squeeze_duration: int, squeeze_avg_vol: float, kcr_val: float) -> np.ndarray | None:
+    if i < 100 or i >= len(df):
+        return None
+
+    o = df["open"].values
+    h = df["high"].values
+    l = df["low"].values
+    c = df["close"].values
+    v = df["volume"].values
+
+    feats = []
+    for w in [12, 24, 48]:
+        if i < w:
+            feats.extend([0] * 12)
+            continue
+
+        win_c = c[i - w : i]
+        win_o = o[i - w : i]
+        win_h = h[i - w : i]
+        win_l = l[i - w : i]
+        win_v = v[i - w : i]
+
+        mn, mx = win_l.min(), max(win_h.max(), win_c.max())
+        rng = mx - mn if mx - mn > 0 else 1e-10
+        total = win_h - win_l
+        total = np.where(total == 0, 1e-10, total)
+        body = np.abs(win_c - win_o) / total
+        direction = np.sign(win_c - win_o)
+        log_c = np.log(np.where(win_c == 0, 1e-10, win_c))
+        rets = np.diff(log_c)
+        v_mean = np.mean(win_v)
+
+        feats.extend([
+            np.mean(rets) if len(rets) > 0 else 0,
+            np.std(rets) if len(rets) > 0 else 0,
+            np.sum(rets) if len(rets) > 0 else 0,
+            float(pd.Series(rets).skew()) if len(rets) > 2 else 0,
+            float(pd.Series(rets).kurtosis()) if len(rets) > 3 else 0,
+            np.mean(body),
+            np.std(body),
+            np.mean(direction),
+            np.mean(direction[-min(3, w):]),
+            (win_c[-1] - mn) / rng,
+            win_v[-1] / v_mean if v_mean > 0 else 1,
+            np.corrcoef(rets[:-1], rets[1:])[0, 1] if len(rets) > 1 and np.std(rets) > 0 else 0,
+        ])
+
+    feats.extend([
+        squeeze_duration,
+        squeeze_duration / (24 * 4),
+        kcr_val,
+        v[i - 1] / squeeze_avg_vol if squeeze_avg_vol > 0 else 1,
+        np.mean(v[max(0, i - 3) : i]) / squeeze_avg_vol if squeeze_avg_vol > 0 else 1,
+    ])
+
+    h48 = np.max(h[max(0, i - 48) : i])
+    l48 = np.min(l[max(0, i - 48) : i])
+    r48 = h48 - l48
+    feats.append((c[i - 1] - l48) / r48 if r48 > 0 else 0.5)
+
+    tr = np.maximum(h[i - 14 : i] - l[i - 14 : i],
+                    np.maximum(np.abs(h[i - 14 : i] - np.roll(c[i - 14 : i], 1)),
+                               np.abs(l[i - 14 : i] - np.roll(c[i - 14 : i], 1))))
+    atr = np.mean(tr[1:])
+    feats.append(atr / c[i - 1] if c[i - 1] > 0 else 0)
+
+    first_ret = (c[i - 1] - c[i - 2]) / c[i - 2] if i >= 2 and c[i - 2] > 0 else 0
+    feats.append(first_ret)
+
+    return np.nan_to_num(np.array(feats), nan=0, posinf=1e6, neginf=-1e6)
+
+
+class SignalEngine:
+    """Rileva squeeze e genera segnali ML in real-time."""
+
+    def __init__(self, bb_w: int = 14, sq_thr: float = 0.8, ml_thr: float = 0.70, min_squeeze_bars: int = 5):
+        self.bb_w = bb_w
+        self.sq_thr = sq_thr
+        self.ml_thr = ml_thr
+        self.min_squeeze_bars = min_squeeze_bars
+
+        self.model: GradientBoostingClassifier | None = None
+        self.scaler: StandardScaler | None = None
+        self.in_squeeze = False
+        self.squeeze_start_idx = 0
+        self.trained = False
+
+    def train(self, df: pd.DataFrame, lookahead: int = 3) -> dict:
+        """Addestra il modello su dati storici."""
+        close = df["close"].values
+        high = df["high"].values
+        low = df["low"].values
+        volume = df["volume"].values
+        n = len(df)
+
+        kcr = keltner_ratio(close, high, low, self.bb_w)
+
+        X_all, y_all = [], []
+        in_sq = False
+        sq_start = 0
+
+        for i in range(self.bb_w + 1, n - lookahead):
+            if np.isnan(kcr[i]):
+                continue
+            is_sq = kcr[i] < self.sq_thr
+            if is_sq and not in_sq:
+                in_sq = True
+                sq_start = i
+            elif not is_sq and in_sq:
+                in_sq = False
+                duration = i - sq_start
+                if duration < self.min_squeeze_bars:
+                    continue
+
+                avg_vol = np.mean(volume[sq_start:i])
+                feats = build_features(df, i, duration, avg_vol, kcr[i])
+                if feats is None:
+                    continue
+
+                actual = (close[i + lookahead - 1] - close[i - 1]) / close[i - 1]
+                X_all.append(feats)
+                y_all.append(1 if actual > 0 else 0)
+
+        if len(X_all) < 30:
+            return {"error": "not enough training samples", "samples": len(X_all)}
+
+        X = np.array(X_all)
+        y = np.array(y_all)
+
+        self.scaler = StandardScaler()
+        X_s = self.scaler.fit_transform(X)
+
+        self.model = GradientBoostingClassifier(
+            n_estimators=150, max_depth=4, min_samples_leaf=10,
+            learning_rate=0.05, subsample=0.8, random_state=42,
+        )
+        self.model.fit(X_s, y)
+        self.trained = True
+
+        preds = self.model.predict(X_s)
+        train_acc = np.mean(preds == y) * 100
+
+        return {"samples": len(X), "up_ratio": np.mean(y) * 100, "train_accuracy": train_acc}
+
+    def check_signal(self, df: pd.DataFrame) -> dict | None:
+        """Controlla se c'è un segnale sulle ultime candele.
+        Ritorna dict con direzione e probabilità, oppure None.
+        """
+        if not self.trained:
+            return None
+
+        close = df["close"].values
+        high = df["high"].values
+        low = df["low"].values
+        volume = df["volume"].values
+        n = len(df)
+
+        kcr = keltner_ratio(close, high, low, self.bb_w)
+
+        if n < self.bb_w + 10:
+            return None
+
+        last_kcr = kcr[-1]
+        prev_kcr = kcr[-2] if n > 1 else np.nan
+
+        if np.isnan(last_kcr) or np.isnan(prev_kcr):
+            return None
+
+        was_squeeze = prev_kcr < self.sq_thr
+        is_released = last_kcr >= self.sq_thr
+
+        if not (was_squeeze and is_released):
+            self.in_squeeze = prev_kcr < self.sq_thr
+            if self.in_squeeze and not hasattr(self, '_sq_start_tracking'):
+                self._sq_start_tracking = n - 1
+            if not self.in_squeeze:
+                self._sq_start_tracking = None
+            return None
+
+        sq_start = getattr(self, '_sq_start_tracking', n - 10)
+        if sq_start is None:
+            sq_start = n - 10
+        duration = (n - 1) - sq_start
+        if duration < self.min_squeeze_bars:
+            self._sq_start_tracking = None
+            return None
+
+        avg_vol = np.mean(volume[max(0, sq_start) : n - 1])
+        feats = build_features(df, n - 1, duration, avg_vol, last_kcr)
+        self._sq_start_tracking = None
+
+        if feats is None:
+            return None
+
+        feats_s = self.scaler.transform(feats.reshape(1, -1))
+        proba = self.model.predict_proba(feats_s)[0]
+        up_idx = list(self.model.classes_).index(1)
+        p_up = proba[up_idx]
+
+        if p_up >= self.ml_thr:
+            return {"direction": "buy", "probability": p_up, "squeeze_duration": duration}
+        elif p_up <= (1 - self.ml_thr):
+            return {"direction": "sell", "probability": 1 - p_up, "squeeze_duration": duration}
+
+        return None