Merge branch 'main' of ssh://git.tielogic.xyz:222/Adriano/PythagorasGoal
# Conflicts: # CLAUDE.md # README.md
This commit is contained in:
Adriano Dal Pastro
@@ -11,6 +11,7 @@ import pandas as pd
|
||||
from src.live.cerbero_client import CerberoClient
|
||||
from src.live.strategy_loader import load_strategy
|
||||
from src.live.strategy_worker import StrategyWorker
|
||||
from src.live.pairs_worker import PairsWorker
|
||||
from src.live.signal_engine import SignalEngine
|
||||
from src.live.telegram_notifier import send_telegram
|
||||
|
||||
@@ -18,9 +19,20 @@ PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||||
DATA_DIR = PROJECT_ROOT / "data" / "paper_trades"
|
||||
|
||||
RESOLUTION_MAP = {"15m": "15", "1h": "60", "5m": "5"}
|
||||
# Convenzione Deribit (verificata via Cerbero, 2026-05-29):
|
||||
# - BTC/ETH = perpetui INVERSE (margine coin): "<COIN>-PERPETUAL"
|
||||
# - altcoin = perpetui LINEARI USDC (margine USDC): "<COIN>_USDC-PERPETUAL", storia dal 2022
|
||||
# Trappola: "LTC-PERPETUAL"/"ADA-PERPETUAL" = 0 candele; "SOL-PERPETUAL" = contratto vecchio
|
||||
# con dati sbagliati. Per gli alt usare SEMPRE la forma _USDC-PERPETUAL.
|
||||
INSTRUMENT_MAP = {
|
||||
"BTC": "BTC-PERPETUAL",
|
||||
"ETH": "ETH-PERPETUAL",
|
||||
"SOL": "SOL_USDC-PERPETUAL",
|
||||
"LTC": "LTC_USDC-PERPETUAL",
|
||||
"ADA": "ADA_USDC-PERPETUAL",
|
||||
"XRP": "XRP_USDC-PERPETUAL",
|
||||
"BNB": "BNB_USDC-PERPETUAL",
|
||||
"DOGE": "DOGE_USDC-PERPETUAL",
|
||||
}
|
||||
|
||||
|
||||
@@ -130,6 +142,26 @@ def build_workers(config: dict) -> tuple[list[StrategyWorker], list[MLWorkerWrap
|
||||
return regular_workers, ml_workers
|
||||
|
||||
|
||||
def build_pairs_workers(config: dict) -> list[PairsWorker]:
|
||||
"""Crea i PairsWorker (2 gambe) dalla sezione `pairs:` dello YAML."""
|
||||
defaults = config.get("defaults", {})
|
||||
workers: list[PairsWorker] = []
|
||||
for entry in config.get("pairs", []):
|
||||
if not entry.get("enabled", True):
|
||||
continue
|
||||
workers.append(PairsWorker(
|
||||
asset_a=entry["a"], asset_b=entry["b"], tf=entry.get("tf", "1h"),
|
||||
params=entry.get("params", {}),
|
||||
capital=entry.get("capital", defaults.get("capital", 1000)),
|
||||
position_size=entry.get("position_size", defaults.get("position_size", 0.15)),
|
||||
leverage=entry.get("leverage", defaults.get("leverage", 3)),
|
||||
fee_rt=entry.get("fee_rt", 0.001),
|
||||
name=entry.get("name", "PR01_pairs_reversion"),
|
||||
data_dir=DATA_DIR,
|
||||
))
|
||||
return workers
|
||||
|
||||
|
||||
def run():
|
||||
config_path = PROJECT_ROOT / "strategies.yml"
|
||||
if not config_path.exists():
|
||||
@@ -143,7 +175,8 @@ def run():
|
||||
train_lookback_days = 365
|
||||
|
||||
regular_workers, ml_workers = build_workers(config)
|
||||
all_worker_count = len(regular_workers) + len(ml_workers)
|
||||
pairs_workers = build_pairs_workers(config)
|
||||
all_worker_count = len(regular_workers) + len(ml_workers) + len(pairs_workers)
|
||||
|
||||
if all_worker_count == 0:
|
||||
print("Nessuna strategia abilitata in strategies.yml")
|
||||
@@ -162,6 +195,8 @@ def run():
|
||||
print(f" • {w.status_summary}")
|
||||
for mw in ml_workers:
|
||||
print(f" • {mw.worker.status_summary} [ML]")
|
||||
for pw in pairs_workers:
|
||||
print(f" • {pw.status_summary} [PAIRS]")
|
||||
|
||||
send_telegram(f"🚀 Multi-Strategy avviato: {all_worker_count} strategie")
|
||||
|
||||
@@ -172,6 +207,9 @@ def run():
|
||||
keys.add((w.asset, w.tf))
|
||||
for mw in ml_workers:
|
||||
keys.add((mw.worker.asset, mw.worker.tf))
|
||||
for pw in pairs_workers: # entrambe le gambe del pair
|
||||
keys.add((pw.asset_a, pw.tf))
|
||||
keys.add((pw.asset_b, pw.tf))
|
||||
return keys
|
||||
|
||||
# Training iniziale ML
|
||||
@@ -253,6 +291,15 @@ def run():
|
||||
except Exception as e:
|
||||
print(f" [{mw.worker.worker_id}] ERRORE: {e}")
|
||||
|
||||
# Tick pairs workers (2 gambe)
|
||||
for pw in pairs_workers:
|
||||
ka, kb = (pw.asset_a, pw.tf), (pw.asset_b, pw.tf)
|
||||
if ka in candle_cache and kb in candle_cache:
|
||||
try:
|
||||
pw.tick(candle_cache[ka], candle_cache[kb])
|
||||
except Exception as e:
|
||||
print(f" [{pw.worker_id}] ERRORE: {e}")
|
||||
|
||||
# Status periodico
|
||||
now = datetime.now(timezone.utc)
|
||||
if now.minute == 0 and now.second < poll_seconds:
|
||||
@@ -261,6 +308,8 @@ def run():
|
||||
lines.append(f" {w.status_summary}")
|
||||
for mw in ml_workers:
|
||||
lines.append(f" {mw.worker.status_summary} [ML]")
|
||||
for pw in pairs_workers:
|
||||
lines.append(f" {pw.status_summary} [PAIRS]")
|
||||
send_telegram("\n".join(lines))
|
||||
|
||||
except KeyboardInterrupt:
|
||||
@@ -277,6 +326,8 @@ def run():
|
||||
if df is not None and not df.empty:
|
||||
mw.worker._close_position(float(df["close"].iloc[-1]), "shutdown")
|
||||
mw.worker._save_state()
|
||||
for pw in pairs_workers: # salva stato; non forzo la chiusura a 2 gambe
|
||||
pw._save_state()
|
||||
send_telegram("🛑 Multi-Strategy arrestato")
|
||||
break
|
||||
except Exception as e:
|
||||
|
||||
@@ -0,0 +1,215 @@
|
||||
"""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}")
|
||||
Reference in New Issue
Block a user