"""Worker per singola strategia — paper trading con stato persistente.""" from __future__ import annotations import json 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.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_trades = 0 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 # 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_trades = state.get("real_trades", 0) 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_trades": self.real_trades, "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) else: self._log("REAL_OPEN_FAIL", {**data, "note": fill.notes}) def _real_close(self, sim_exit: float, reason: str, sim_pnl: float): """Chiusura REALE (reduce-only della quota worker) + confronto col sim.""" if not self.real_in_position: return fill = self.executor.close_amount(self.exec_instrument, self.real_side, self.real_amount, label=self.worker_id) exit_price = fill.fill_price or sim_exit 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 + fill.fee_usd 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, "sim_exit": round(sim_exit, 2), "real_fill": fill.fill_price, "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(fill.fee_usd, 5), "real_pnl_usd": round(real_pnl, 4), "sim_pnl_usd": round(sim_pnl, 4), "real_capital": round(self.real_capital, 4), "verified": fill.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 = "" 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: # 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}")