"""PairsWorker — paper trading a 2 GAMBE per la famiglia PR01 (spread reversion). Market-neutral: long asset A / short asset B (o viceversa) sullo z-score del log-ratio. Distinto dallo StrategyWorker single-leg: gestisce due strumenti, due prezzi di ingresso, e conta le fee su ENTRAMBE le gambe (2*fee_rt*lev = 0.20% RT/coppia con fee_rt=0.001). Semantica identica al backtest scripts/analysis/pairs_research.pairs_sim: r[i] = log(closeA[i]/closeB[i]); z[i] = (r[i]-SMA_n(r)[i]) / STD_n(r)[i] (causale) ENTRY a close[i]: z<=-z_in -> LONG ratio (long A / short B); z>=+z_in -> SHORT ratio EXIT: |z| <= z_exit (rientro) oppure time-limit max_bars filtro candele sporche: salta l'ingresso se |dr[i]| > jump_max PnL = (retA - retB) * direction * lev - 2*fee_rt*lev (notional uguale per gamba) Stato persistente (resume al restart) e log come StrategyWorker. """ 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.live.telegram_notifier import notify_event class PairsWorker: def __init__( self, asset_a: str, asset_b: str, tf: str, params: dict | None = None, capital: float = 1000.0, position_size: float = 0.15, leverage: float = 3.0, fee_rt: float = 0.001, # per gamba RT; la coppia paga 2x name: str = "PR01_pairs_reversion", data_dir: Path = Path("data/paper_trades"), ): self.asset_a = asset_a self.asset_b = asset_b self.tf = tf self.name = name p = params or {} self.n = int(p.get("n", 50)) self.z_in = float(p.get("z_in", 2.0)) self.z_exit = float(p.get("z_exit", 0.75)) self.max_bars = int(p.get("max_bars", 72)) self.jump_max = float(p.get("jump_max", 0.08)) self.initial_capital = capital self.position_size = position_size self.leverage = leverage self.fee_rt = fee_rt self.worker_id = f"{name}__{asset_a}_{asset_b}__{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 = 0 # +1 long ratio (long A/short B), -1 short ratio self.entry_a = 0.0 self.entry_b = 0.0 self.entry_z = 0.0 self.entry_time = "" self.bars_held = 0 self.total_trades = 0 self.total_wins = 0 self.last_bar_ts = 0 self.started_at = datetime.now(timezone.utc).isoformat() self._load_state() self._save_state() # ---------------- persistenza ---------------- def _load_state(self): if not self.status_path.exists(): self._log("INIT", {"capital": self.capital, "pair": f"{self.asset_a}/{self.asset_b}", "tf": self.tf, "params": {"n": self.n, "z_in": self.z_in, "z_exit": self.z_exit, "max_bars": self.max_bars}}) return with open(self.status_path) as f: s = json.load(f) self.capital = s.get("capital", self.initial_capital) self.in_position = s.get("in_position", False) self.direction = s.get("direction", 0) self.entry_a = s.get("entry_a", 0.0) self.entry_b = s.get("entry_b", 0.0) self.entry_z = s.get("entry_z", 0.0) self.entry_time = s.get("entry_time", "") self.bars_held = s.get("bars_held", 0) self.total_trades = s.get("total_trades", 0) self.total_wins = s.get("total_wins", 0) self.last_bar_ts = s.get("last_bar_ts", 0) self.started_at = s.get("started_at", self.started_at) self._log("RESUME", {"capital": round(self.capital, 2), "total_trades": self.total_trades, "in_position": self.in_position}) def _save_state(self): state = { "capital": round(self.capital, 2), "in_position": self.in_position, "direction": self.direction, "entry_a": self.entry_a, "entry_b": self.entry_b, "entry_z": round(self.entry_z, 4), "entry_time": self.entry_time, "bars_held": self.bars_held, "total_trades": self.total_trades, "total_wins": self.total_wins, "last_bar_ts": self.last_bar_ts, "started_at": self.started_at, "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): notify_event(event, {"worker": self.worker_id, **(data or {})}) # ---------------- segnale ---------------- def _zscore(self, ca: np.ndarray, cb: np.ndarray) -> tuple[np.ndarray, np.ndarray]: r = np.log(ca / cb) ma = pd.Series(r).rolling(self.n).mean().values sd = pd.Series(r).rolling(self.n).std().values z = (r - ma) / np.where(sd == 0, np.nan, sd) dr = np.abs(np.diff(r, prepend=r[0])) return z, dr # ---------------- trading ---------------- def _open(self, d: int, ca: float, cb: float, z: float): self.in_position = True self.direction = d self.entry_a, self.entry_b, self.entry_z = ca, cb, z self.entry_time = datetime.now(timezone.utc).isoformat() self.bars_held = 0 data = {"direction": "long_ratio" if d == 1 else "short_ratio", "long_leg": self.asset_a if d == 1 else self.asset_b, "short_leg": self.asset_b if d == 1 else self.asset_a, "entry_a": round(ca, 4), "entry_b": round(cb, 4), "z": round(z, 3), "capital": round(self.capital, 2)} self._log("OPEN", data); self._notify("OPENED", data) def _close(self, ca: float, cb: float, z: float, reason: str): if not self.in_position: return ret_a = (ca - self.entry_a) / self.entry_a ret_b = (cb - self.entry_b) / self.entry_b gross = (ret_a - ret_b) * self.direction * self.leverage fee = 2 * self.fee_rt * self.leverage # 2 gambe net = gross - fee pnl = self.capital * self.position_size * net self.capital = max(self.capital + pnl, 0.0) is_win = net > 0 self.total_trades += 1 self.total_wins += is_win acc = self.total_wins / self.total_trades * 100 if self.total_trades else 0 data = {"reason": reason, "exit_a": round(ca, 4), "exit_b": round(cb, 4), "z": round(z, 3), "gross_ret": round(gross * 100, 3), "fee": round(fee * 100, 3), "net_return": round(net * 100, 3), "pnl": round(pnl, 2), "capital": round(self.capital, 2), "bars_held": self.bars_held, "win": bool(is_win), "total_trades": self.total_trades, "accuracy": round(acc, 1)} self._log("CLOSE", data); self._notify("CLOSED", data) self.in_position = False self.direction = 0 self.entry_a = self.entry_b = self.entry_z = 0.0 self.bars_held = 0 def tick(self, df_a: pd.DataFrame, df_b: pd.DataFrame): """Chiamato ad ogni poll con gli OHLCV aggiornati delle due gambe.""" if df_a is None or df_b is None or df_a.empty or df_b.empty: return m = df_a[["timestamp", "close"]].rename(columns={"close": "ca"}).merge( df_b[["timestamp", "close"]].rename(columns={"close": "cb"}), on="timestamp", how="inner" ).sort_values("timestamp").reset_index(drop=True) if len(m) < self.n + 2: return ca, cb = m["ca"].values, m["cb"].values z, dr = self._zscore(ca, cb) i = len(m) - 1 cur_ts = int(m["timestamp"].iloc[i]) zi = z[i] if np.isnan(zi): self._save_state(); return if self.in_position: if cur_ts > self.last_bar_ts: self.bars_held += 1 self.last_bar_ts = cur_ts if abs(zi) <= self.z_exit: self._close(float(ca[i]), float(cb[i]), float(zi), "mean_revert") elif self.bars_held >= self.max_bars: self._close(float(ca[i]), float(cb[i]), float(zi), "time_limit") self._save_state() return # flat: cerca ingresso (no look-ahead: z[i] usa solo dati <= i) if dr[i] <= self.jump_max: if zi <= -self.z_in: self._open(1, float(ca[i]), float(cb[i]), float(zi)); self.last_bar_ts = cur_ts elif zi >= self.z_in: self._open(-1, float(ca[i]), float(cb[i]), float(zi)); self.last_bar_ts = cur_ts self._save_state() @property def status_summary(self) -> str: acc = self.total_wins / self.total_trades * 100 if self.total_trades else 0 pos = ("LONG " + self.asset_a if self.direction == 1 else "SHORT " + self.asset_a if self.direction == -1 else "FLAT") return (f"{self.worker_id}: €{self.capital:.0f} | {self.total_trades}t {acc:.0f}% | {pos}")