"""Worker per singola strategia — paper trading con stato persistente.""" from __future__ import annotations import json import time from datetime import datetime, timezone from pathlib import Path import numpy as np import pandas as pd from src.strategies.base import Strategy, Signal from src.strategies.fade_base import atr as _atr from src.live.telegram_notifier import notify_event from src.live.execution import ExecutionClient FEE_RT = 0.002 class StrategyWorker: """Gestisce paper trading per una singola strategia/asset/tf.""" def __init__( self, strategy: Strategy, asset: str, tf: str, capital: float = 1000.0, position_size: float = 0.15, leverage: float = 3.0, hold_bars: int = 3, params: dict | None = None, data_dir: Path = Path("data/paper_trades"), executor: ExecutionClient | None = None, exec_instrument: str | None = None, ): self.strategy = strategy self.asset = asset self.tf = tf self.initial_capital = capital self.position_size = position_size self.leverage = leverage self.hold_bars = hold_bars self.params = params or {} # --- Esecuzione REALE (shadow): se attiva, ogni open/close sim e' affiancato # da un ordine reale su Deribit (lineare USDC), con ledger reale parallelo. --- self.executor = executor self.exec_instrument = exec_instrument self.execution_enabled = bool(executor and exec_instrument) self.real_capital = capital self.real_in_position = False self.real_side = "" # "buy" | "sell" dell'apertura reale self.real_amount = 0.0 # amount Deribit (base-coin) da richiudere self.real_entry_price = 0.0 self.real_entry_fee_usd = 0.0 self.real_entry_notional = 0.0 # USD effettivi esposti all'entrata self.real_order_id = "" self.real_tp_order_id = "" # LIMIT reduce-only resting al TP (persistito per il resume) self.real_trades = 0 self.real_first_notified = False # alert Telegram "esecuzione viva" una tantum self.worker_id = f"{strategy.name}__{asset}__{tf}" self.work_dir = data_dir / self.worker_id self.work_dir.mkdir(parents=True, exist_ok=True) self.trades_path = self.work_dir / "trades.jsonl" self.status_path = self.work_dir / "status.json" self.capital = capital self.in_position = False self.direction: int = 0 self.entry_price: float = 0 self.entry_time: str = "" self.bars_held: int = 0 self.total_trades: int = 0 self.total_wins: int = 0 self.started_at = datetime.now(timezone.utc).isoformat() self.last_bar_ts: int = 0 # Exit guidati dalla strategia via Signal.metadata (0 = usa hold_bars/stop legacy) self.tp: float = 0.0 self.sl: float = 0.0 self.max_bars: int = 0 # EXIT-16 close-confirm SL (2026-06-04, fade): se settato nei params dello # sleeve, lo SL intrabar e' disattivato e lo stop scatta solo se il CLOSE # sfonda sl di sl_confirm_atr*ATR14 (immune ai wick). TP intrabar invariato. self.sl_confirm_atr: float | None = ( float(self.params["sl_confirm_atr"]) if self.params.get("sl_confirm_atr") else None) # Fee dalla strategia (MR01 = 0.001 realistico Deribit), fallback al default modulo self.fee_rt: float = float(getattr(strategy, "fee_rt", FEE_RT)) self._load_state() self._save_state() def _load_state(self): """Riprende stato da status.json se esiste.""" if not self.status_path.exists(): self._log("INIT", {"capital": self.capital, "strategy": self.strategy.name, "asset": self.asset, "tf": self.tf}) return with open(self.status_path) as f: state = json.load(f) self.capital = state.get("capital", self.initial_capital) self.in_position = state.get("in_position", False) self.direction = state.get("direction", 0) self.entry_price = state.get("entry_price", 0) self.entry_time = state.get("entry_time", "") self.bars_held = state.get("bars_held", 0) self.total_trades = state.get("total_trades", 0) self.total_wins = state.get("total_wins", 0) self.started_at = state.get("started_at", self.started_at) self.last_bar_ts = state.get("last_bar_ts", 0) self.tp = state.get("tp", 0.0) self.sl = state.get("sl", 0.0) self.max_bars = state.get("max_bars", 0) self.real_capital = state.get("real_capital", self.initial_capital) self.real_in_position = state.get("real_in_position", False) self.real_side = state.get("real_side", "") self.real_amount = state.get("real_amount", 0.0) self.real_entry_price = state.get("real_entry_price", 0.0) self.real_entry_fee_usd = state.get("real_entry_fee_usd", 0.0) self.real_entry_notional = state.get("real_entry_notional", 0.0) self.real_order_id = state.get("real_order_id", "") self.real_tp_order_id = state.get("real_tp_order_id", "") self.real_trades = state.get("real_trades", 0) self.real_first_notified = state.get("real_first_notified", False) self._log("RESUME", {"capital": round(self.capital, 2), "total_trades": self.total_trades, "in_position": self.in_position, "real_capital": round(self.real_capital, 2), "real_in_position": self.real_in_position}) def _save_state(self): state = { "capital": round(self.capital, 2), "in_position": self.in_position, "direction": self.direction, "entry_price": self.entry_price, "entry_time": self.entry_time, "bars_held": self.bars_held, "total_trades": self.total_trades, "total_wins": self.total_wins, "started_at": self.started_at, "last_bar_ts": self.last_bar_ts, "tp": self.tp, "sl": self.sl, "max_bars": self.max_bars, "real_capital": round(self.real_capital, 4), "real_in_position": self.real_in_position, "real_side": self.real_side, "real_amount": self.real_amount, "real_entry_price": self.real_entry_price, "real_entry_fee_usd": self.real_entry_fee_usd, "real_entry_notional": self.real_entry_notional, "real_order_id": self.real_order_id, "real_tp_order_id": self.real_tp_order_id, "real_trades": self.real_trades, "real_first_notified": self.real_first_notified, "last_update": datetime.now(timezone.utc).isoformat(), } with open(self.status_path, "w") as f: json.dump(state, f, indent=2) def _log(self, event: str, data: dict | None = None): entry = { "ts": datetime.now(timezone.utc).isoformat(), "worker": self.worker_id, "event": event, **(data or {}), } with open(self.trades_path, "a") as f: f.write(json.dumps(entry) + "\n") print(f" [{self.worker_id}] {event}: {json.dumps(data or {}, default=str)}") def _notify(self, event: str, data: dict | None = None): enriched = {"worker": self.worker_id, **(data or {})} notify_event(event, enriched) def _open_position(self, signal: Signal, current_price: float): notional = self.capital * self.position_size * self.leverage size = notional / current_price if current_price > 0 else 0 self.in_position = True self.direction = signal.direction self.entry_price = current_price self.entry_time = datetime.now(timezone.utc).isoformat() self.bars_held = 0 meta = signal.metadata or {} self.tp = float(meta.get("tp", 0.0) or 0.0) self.sl = float(meta.get("sl", 0.0) or 0.0) self.max_bars = int(meta.get("max_bars", 0) or 0) trade_data = { "direction": "long" if signal.direction == 1 else "short", "price": round(current_price, 2), "size": round(size, 6), "notional": round(notional, 2), "capital": round(self.capital, 2), "tp": round(self.tp, 2) if self.tp else None, "sl": round(self.sl, 2) if self.sl else None, } self._log("OPEN", trade_data) self._notify("OPENED", trade_data) if self.execution_enabled: self._real_open(signal.direction, current_price, notional) def _real_open(self, direction: int, sim_price: float, notional: float): """Apertura REALE (shadow) accanto al fill simulato. Logga il confronto prezzo-sim vs prezzo-eseguito e la fee reale Deribit.""" from src.live.execution import contract_spec side = "buy" if direction == 1 else "sell" fill = self.executor.open(self.exec_instrument, side, notional, label=self.worker_id) slip_bps = ((fill.fill_price / sim_price - 1) * 1e4 if fill.fill_price and sim_price else None) data = { "instrument": self.exec_instrument, "side": side, "order_id": fill.order_id, "amount": fill.amount, "sim_price": round(sim_price, 2), "real_fill": fill.fill_price, "slippage_bps": round(slip_bps, 2) if slip_bps is not None else None, "fee_usd": round(fill.fee_usd, 5), "verified": fill.verified, } if fill.verified: linear = contract_spec(self.exec_instrument).get("linear") self.real_in_position = True self.real_side = side self.real_amount = fill.amount self.real_entry_price = fill.fill_price or sim_price self.real_entry_fee_usd = fill.fee_usd self.real_entry_notional = (fill.amount * self.real_entry_price if linear else fill.amount) self.real_order_id = fill.order_id or "" self._log("REAL_OPEN", data) if not self.real_first_notified: # conferma una-tantum: l'esecuzione reale e' viva self._notify("REAL_EXEC_LIVE", data) self.real_first_notified = True self._place_real_tp() else: self._log("REAL_OPEN_FAIL", {**data, "note": fill.notes}) self._notify("REAL_OPEN_FAIL", {**data, "note": fill.notes}) def _place_real_tp(self): """LIMIT reduce-only appoggiato al TP della strategia (fix divergenza sim/reale 2026-06-04: il market-on-poll usciva post-rimbalzo, +235 bps sopra il livello TP). Copre la SOLA quota del worker. Se il piazzamento fallisce si resta sul fallback market-on-poll di _real_close.""" self.real_tp_order_id = "" if not (self.tp and self.real_amount > 0): return rest = self.executor.place_tp_limit(self.exec_instrument, self.real_side, self.real_amount, self.tp, label=self.worker_id) data = { "instrument": self.exec_instrument, "order_id": rest.order_id, "tp": round(self.tp, 2), "amount": self.real_amount, "state": rest.order_state, } if rest.verified and rest.order_id: self.real_tp_order_id = rest.order_id self._log("REAL_TP_RESTING", data) else: self._log("REAL_TP_FAIL", {**data, "note": rest.notes}) def _real_close(self, sim_exit: float, reason: str, sim_pnl: float): """Chiusura REALE (reduce-only della quota worker) + confronto col sim. Prima riconcilia l'eventuale LIMIT resting al TP: lo cancella (innocuo se gia' fillato — cosi' nessun fill puo' arrivare DOPO la lettura) e legge i fill reali dal trade history per order_id; solo la quota residua viene chiusa a mercato (fallback, o exit non-TP: stop-loss/time_limit). L'uscita take-profit reale avviene cosi' AL livello come nel backtest, non al poll post-rimbalzo.""" if not self.real_in_position: return from src.live.execution import contract_spec step = contract_spec(self.exec_instrument)["step"] # 1) ordine TP resting: cancella, poi riconcilia i fill (order_id su history) tp_amt, tp_px, tp_fee = 0.0, None, 0.0 tp_order_id = self.real_tp_order_id if tp_order_id: cres = self.executor.cancel_order(tp_order_id) cancelled = cres.get("state") == "cancelled" for _ in range(self.executor.verify_polls): tp_amt, tp_px, tp_fee = self.executor.resting_fills( self.exec_instrument, tp_order_id) if tp_amt > 0 or cancelled: break # cancel pulito = al piu' fill parziali gia' visti time.sleep(self.executor.verify_sleep) tp_amt = min(tp_amt, self.real_amount) if tp_amt > 0 and not tp_px: tp_px = self.tp or sim_exit # fallback: il limit filla al suo livello # 2) quota residua → market reduce-only (mai close_position: strumento condiviso) remainder = self.real_amount - tp_amt fill = None if remainder >= step / 2: fill = self.executor.close_amount(self.exec_instrument, self.real_side, remainder, label=self.worker_id) market_amt = fill.amount if (fill and fill.verified) else 0.0 # 3) prezzo d'uscita combinato (media pesata TP-fill + market) e fee totali parts = [(a, p) for a, p in ((tp_amt, tp_px), (market_amt, fill.fill_price if fill else None)) if a > 0 and p] exit_price = (sum(a * p for a, p in parts) / sum(a for a, _ in parts) if parts else sim_exit) exit_fee = tp_fee + (fill.fee_usd if fill else 0.0) verified = (tp_amt + market_amt) >= self.real_amount - step / 2 rdir = 1 if self.real_side == "buy" else -1 price_change = (exit_price - self.real_entry_price) / self.real_entry_price \ if self.real_entry_price else 0.0 real_gross = rdir * price_change * self.real_entry_notional real_fees = self.real_entry_fee_usd + exit_fee real_pnl = real_gross - real_fees self.real_capital += real_pnl self.real_trades += 1 slip_bps = ((exit_price / sim_exit - 1) * 1e4 if exit_price and sim_exit else None) self._log("REAL_CLOSE", { "reason": reason, "order_id": fill.order_id if fill else tp_order_id, "tp_order_id": tp_order_id or None, "tp_filled_amount": tp_amt, "market_amount": market_amt, "sim_exit": round(sim_exit, 2), "real_fill": round(exit_price, 2) if parts else None, "slippage_bps": round(slip_bps, 2) if slip_bps is not None else None, "entry_fee_usd": round(self.real_entry_fee_usd, 5), "exit_fee_usd": round(exit_fee, 5), "real_pnl_usd": round(real_pnl, 4), "sim_pnl_usd": round(sim_pnl, 4), "real_capital": round(self.real_capital, 4), "verified": verified, }) self.real_in_position = False self.real_side = "" self.real_amount = 0.0 self.real_entry_price = 0.0 self.real_entry_fee_usd = 0.0 self.real_entry_notional = 0.0 self.real_order_id = "" self.real_tp_order_id = "" def _close_position(self, current_price: float, reason: str): if not self.in_position: return price_change = (current_price - self.entry_price) / self.entry_price trade_return = price_change * self.direction net = trade_return * self.leverage - self.fee_rt * self.leverage pnl = self.capital * self.position_size * net is_win = net > 0 # win = profitto NETTO dopo fee (non il lordo trade_return) self.capital += pnl self.capital = max(self.capital, 0) self.total_trades += 1 if is_win: self.total_wins += 1 accuracy = self.total_wins / self.total_trades * 100 if self.total_trades > 0 else 0 trade_data = { "reason": reason, "direction": "long" if self.direction == 1 else "short", "entry": round(self.entry_price, 2), "exit": round(current_price, 2), "pnl": round(pnl, 2), "net_return": round(net * 100, 3), "capital": round(self.capital, 2), "bars_held": self.bars_held, "win": is_win, "total_trades": self.total_trades, "accuracy": round(accuracy, 1), } self._log("CLOSE", trade_data) self._notify("CLOSED", trade_data) if self.execution_enabled: self._real_close(current_price, reason, pnl) self.in_position = False self.direction = 0 self.entry_price = 0 self.entry_time = "" self.bars_held = 0 self.tp = 0.0 self.sl = 0.0 self.max_bars = 0 def tick(self, df: pd.DataFrame, df_1h: pd.DataFrame | None = None): """Chiamato ad ogni poll con DataFrame OHLCV aggiornato. df_1h: serie 1h live opzionale per strategie multi-timeframe (es. MT01), passata ai generate_signals via params. Se None la strategia ricade sul parquet statico. """ if df.empty or len(df) < 100: return c = df["close"].values current_price = float(c[-1]) bar_high = float(df["high"].iloc[-1]) bar_low = float(df["low"].iloc[-1]) current_ts = int(df["timestamp"].iloc[-1]) ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) if self.in_position: if current_ts > self.last_bar_ts: self.bars_held += 1 self.last_bar_ts = current_ts if self.tp and self.sl and self.sl_confirm_atr: # EXIT-16 close-confirm (2026-06-04): TP intrabar al livello come il # backtest; lo SL scatta SOLO se il close sfonda sl ∓ buf*ATR14 — i # wick che bucano lo stop e rientrano (l'overshoot che la fade fada) # non stoppano piu'. PORT06: OOS Sharpe 8.82->10.06 (exit-lab, 34 agenti). # # FIX 2026-06-05: il confirm va valutato sul close di barra COMPLETATA, # come nel backtest (fade_base: c[j] di bar chiusi) — NON sul prezzo # della barra in formazione, che reintroduce la wick-sensitivity che # EXIT-16 elimina (audit live: 2 stop su 3 del 2026-06-05 erano scattati # su dip intrabar che il backtest avrebbe ignorato in quel momento). # L'ultima riga del df e' la candela in corso se non e' ancora trascorsa # la sua durata; il fill resta al prezzo corrente (lag di poll, stress # lag_close_exit superato in exit-lab). Il buf usa l'ATR della stessa # barra completata. ts_arr = df["timestamp"].values.astype("int64") bar_ms = int(np.median(np.diff(ts_arr[-50:]))) if len(ts_arr) > 1 else 0 now_ms = int(time.time() * 1000) k = -1 if now_ms >= ts_arr[-1] + bar_ms else -2 confirm_close = float(c[k]) buf = self.sl_confirm_atr * float(_atr(df, 14)[k]) if not np.isfinite(buf): buf = 0.0 if self.direction == 1: if bar_high >= self.tp: self._close_position(self.tp, "take_profit") elif confirm_close < self.sl - buf: self._close_position(current_price, "stop_loss") elif self.max_bars and self.bars_held >= self.max_bars: self._close_position(current_price, "time_limit") else: if bar_low <= self.tp: self._close_position(self.tp, "take_profit") elif confirm_close > self.sl + buf: self._close_position(current_price, "stop_loss") elif self.max_bars and self.bars_held >= self.max_bars: self._close_position(current_price, "time_limit") elif self.tp and self.sl: # Exit INTRABAR come il backtest: si controllano high/low della barra (non solo il # close) e si esce AL LIVELLO tp/sl. SL prima (conservativo), poi TP, poi time-limit. if self.direction == 1: if bar_low <= self.sl: self._close_position(self.sl, "stop_loss") elif bar_high >= self.tp: self._close_position(self.tp, "take_profit") elif self.max_bars and self.bars_held >= self.max_bars: self._close_position(current_price, "time_limit") else: if bar_high >= self.sl: self._close_position(self.sl, "stop_loss") elif bar_low <= self.tp: self._close_position(self.tp, "take_profit") elif self.max_bars and self.bars_held >= self.max_bars: self._close_position(current_price, "time_limit") elif self.max_bars: # Exit puro a orizzonte (strategie senza TP/SL, es. SH01 shape-ML H=12): # onora max_bars dalla metadata del Signal, non il fallback hold_bars=3. if self.bars_held >= self.max_bars: self._close_position(current_price, "time_limit") elif self.bars_held >= self.hold_bars: self._close_position(current_price, "hold_limit") else: pnl_pct = (current_price - self.entry_price) / self.entry_price * self.direction if pnl_pct <= -0.02: self._close_position(current_price, "stop_loss") self._save_state() return # Genera segnali extra = dict(self.params) if df_1h is not None: extra["df_1h"] = df_1h signals = self.strategy.generate_signals( df, ts, asset=self.asset, tf=self.tf, **extra ) if not signals: self._save_state() return last_signal = signals[-1] last_idx = len(df) - 1 if last_signal.idx >= last_idx - 1: self._open_position(last_signal, current_price) self.last_bar_ts = current_ts self._save_state() @property def status_summary(self) -> str: acc = self.total_wins / self.total_trades * 100 if self.total_trades > 0 else 0 pos = "LONG" if self.direction == 1 else "SHORT" if self.direction == -1 else "FLAT" return (f"{self.worker_id}: €{self.capital:.0f} | {self.total_trades}t " f"{acc:.0f}% | {pos}")