3 Commits

Author SHA1 Message Date
Adriano a60ad30ac0 fix(live): naming Deribit corretto per alt -> tutte le 5 coppie pairs tradabili live
Gli alt su Deribit sono perp LINEARI USDC: "<COIN>_USDC-PERPETUAL" (storia dal 2022),
non "<COIN>-PERPETUAL" (vuoto per LTC/ADA, dati errati per SOL). INSTRUMENT_MAP corretto.
Smoke test live (live_smoke_pairs.py): tutte e 5 le coppie ricevono feed fresco (1448
barre, ultima ~0.4h) e ticcano. Riabilitate tutte le coppie in strategies.yml.
BTC/ETH restano inverse ("<COIN>-PERPETUAL"). CLAUDE.md / docstring PR01 aggiornati.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-29 09:26:27 +02:00
Adriano bd31a15548 fix(live): smoke test REALE pairs -> live solo ETH/BTC (alt assenti su Deribit)
Test reale (scripts/analysis/live_smoke_pairs.py): fetch live Cerbero + tick vero per
coppia. Scoperta: l'endpoint Deribit serve solo BTC/ETH freschi; LTC/ADA-PERPETUAL sono
VUOTI e SOL ha pochi dati. Il backtest usava i parquet locali (8 asset completi), ma la
pipeline live no -> tradabile live SOLO ETH/BTC.

- strategies.yml: abilitata solo la coppia ETH/BTC; le altre 4 disabilitate con nota
  (valide a backtest, off finche' non si aggiunge un feed live per gli alt).
- live_smoke_pairs.py (nuovo): verifica end-to-end della pipeline live (no ordini reali).
- CLAUDE.md / docstring PR01: distinzione esplicita logica-validata vs live-disponibile.

Onesta': la validazione precedente era backtest-equivalence (ESATTA), NON test live;
il test live ha rivelato il limite del feed alt.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-29 09:16:50 +02:00
Adriano 4dc0e77ee5 feat(live): worker a 2 gambe per i pairs (PR01 market-neutral)
src/live/pairs_worker.py: PairsWorker market-neutral (long A / short B sullo z-score
del log-ratio, exit |z|<=z_exit o max_bars, FEE SU 2 GAMBE = 2*fee_rt*lev, stato
persistente come StrategyWorker). multi_runner: sezione `pairs:` nello YAML, fetch di
entrambe le gambe, tick/status/shutdown; INSTRUMENT_MAP esteso agli alt. strategies.yml:
5 coppie PR01 (config universale n50 z2 zx0.75 mb72).

Validazione (scripts/analysis/validate_worker_pairs.py): replay live bar-per-bar ==
backtest pairs_sim ESATTAMENTE -> ETH/BTC capitale 2.870.429 = 2.870.429, 1754 trade,
win 74.1% identici. Caveat: shortabilita'/liquidita' del perp B sugli alt da verificare
in trading reale. CLAUDE.md / docstring PR01 aggiornati.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-29 09:12:25 +02:00
7 changed files with 492 additions and 10 deletions
+15 -5
View File
@@ -24,10 +24,12 @@ src/strategies/ → classe base Strategy ABC + indicatori condivisi
base.py → Strategy, Signal, BacktestResult, YearlyStats base.py → Strategy, Signal, BacktestResult, YearlyStats
indicators.py → keltner_ratio, detect_squeezes, ema, atr, rv, correlation indicators.py → keltner_ratio, detect_squeezes, ema, atr, rv, correlation
src/live/ → paper trading live multi-strategia src/live/ → paper trading live multi-strategia
multi_runner.py → orchestratore: carica YAML, fetch candele, tick worker multi_runner.py → orchestratore: carica YAML (strategies + pairs), fetch candele, tick worker
strategy_worker.py → worker indipendente: capital, trade log, stato persistente. strategy_worker.py → worker single-leg: capital, trade log, stato persistente.
Exit guidati da strategia (TP/SL/max_bars via Signal.metadata), Exit guidati da strategia (TP/SL/max_bars via Signal.metadata),
fallback hold_bars/stop -2%. Usa fee_rt della strategia. fallback hold_bars/stop -2%. Usa fee_rt della strategia.
pairs_worker.py → worker a 2 GAMBE per PR01 (market-neutral): long A / short B sullo
z-score del log-ratio, exit |z|<=z_exit o max_bars, fee su 2 gambe.
strategy_loader.py → import dinamico classi Strategy da scripts/strategies/ strategy_loader.py → import dinamico classi Strategy da scripts/strategies/
cerbero_client.py → client HTTP per Cerbero MCP (Deribit testnet) cerbero_client.py → client HTTP per Cerbero MCP (Deribit testnet)
signal_engine.py → squeeze + ML real-time (legacy ML01, ora in waste) + validazione OOS signal_engine.py → squeeze + ML real-time (legacy ML01, ora in waste) + validazione OOS
@@ -157,7 +159,14 @@ quest'ultima riconferma la dominanza mean-reversion). Due edge reali:
(Sharpe 4.36), LTC/ETH (3.08), ADA/ETH (2.69), BTC/LTC (2.36, robusta anche 4h), ETH/SOL (Sharpe 4.36), LTC/ETH (3.08), ADA/ETH (2.69), BTC/LTC (2.36, robusta anche 4h), ETH/SOL
(1.96, la più debole). Pattern: sempre alt-liquido vs major. Plateau confermato (1.96, la più debole). Pattern: sempre alt-liquido vs major. Plateau confermato
(heatmap 20/20 Sharpe>1) + walk-forward (ETH/BTC 11/12 finestre+). **BNB/ETH scartata** (heatmap 20/20 Sharpe>1) + walk-forward (ETH/BTC 11/12 finestre+). **BNB/ETH scartata**
(overfit). Corr col mercato ~0.02-0.08. Fee su **2 gambe** (worker da estendere). Verifica: `pairs_research.py`. (overfit). Corr col mercato ~0.02-0.08. Fee su **2 gambe**: worker live implementato
(`src/live/pairs_worker.py`, sezione `pairs:` in `strategies.yml`). LOGICA validata
(`validate_worker_pairs.py`: replay == backtest ESATTO). LIVE (`live_smoke_pairs.py`,
smoke reale Cerbero): **tutte e 5 le coppie con feed live fresco**. Naming Deribit:
BTC/ETH = `<COIN>-PERPETUAL` (inverse); alt = `<COIN>_USDC-PERPETUAL` (lineari USDC,
storia dal 2022). Trappola: `LTC-PERPETUAL`/`SOL-PERPETUAL` danno vuoto/dati errati →
usare sempre `_USDC-PERPETUAL`. Resta da verificare solo liquidità/fill in esecuzione.
Verifica edge: `pairs_research.py`.
- **TSM01** (`scripts/analysis/tsmom_research.py`): TSMOM multi-orizzonte 3/6/12m + risk-off, - **TSM01** (`scripts/analysis/tsmom_research.py`): TSMOM multi-orizzonte 3/6/12m + risk-off,
**gross 0.30**, distinto da ROT02 (corr 0.62), DD 15-22%, mai un anno negativo. Robusto **gross 0.30**, distinto da ROT02 (corr 0.62), DD 15-22%, mai un anno negativo. Robusto
(36/36 config OOS+) ma diversificatore, non motore di ritorno (rende meno di ROT02). (36/36 config OOS+) ma diversificatore, non motore di ritorno (rende meno di ROT02).
@@ -167,8 +176,9 @@ grande (`scripts/analysis/combine_v2.py`). **Numeri sobri onesti** (l'OOS singol
è regime calmo → ottimistico ~50%): worst-DD su 90g rolling **~6%** (non 2.3%), Sharpe è regime calmo → ottimistico ~50%): worst-DD su 90g rolling **~6%** (non 2.3%), Sharpe
atteso **~5** (mediana semestrale), ogni anno positivo dal 2021, regge **leva 2x + atteso **~5** (mediana semestrale), ogni anno positivo dal 2021, regge **leva 2x +
slippage doppio** (CAGR 36%, Sharpe 5.1). Config robusta raccomandata: **MASTER-esteso slippage doppio** (CAGR 36%, Sharpe 5.1). Config robusta raccomandata: **MASTER-esteso
equal-weight, leva 2x, cap pairs ~30-35%** (i pairs sono ~57% del rischio e non ancora equal-weight, leva 2x, cap pairs ~30-35%** (i pairs sono ~57% del rischio; worker live a
validati col worker live a 2 gambe). La confluenza multi-TF è stata SCARTATA (overfit). 2 gambe implementato, validato e con feed live su tutte e 5 le coppie — resta da
verificare liquidità/fill in esecuzione reale). La confluenza multi-TF è stata SCARTATA (overfit).
**Metodologia obbligatoria per ogni nuova strategia** (per non ripetere l'errore squeeze): **Metodologia obbligatoria per ogni nuova strategia** (per non ripetere l'errore squeeze):
1. Ingresso eseguibile: direzione e prezzo decisi con dati **fino a `close[i]`**, mai `close[i-1]` con direzione da `i`. 1. Ingresso eseguibile: direzione e prezzo decisi con dati **fino a `close[i]`**, mai `close[i-1]` con direzione da `i`.
+87
View File
@@ -0,0 +1,87 @@
"""Smoke test REALE dei pairs: fetch live da Cerbero + un tick vero per coppia.
A differenza di validate_worker_pairs.py (replay su parquet storici), questo verifica
la PIPELINE LIVE end-to-end: chiama Cerbero per entrambe le gambe, controlla che lo
strumento esista e sia fresco, fa un tick reale del PairsWorker e riporta lo stato.
Serve a scoprire i problemi che il backtest nasconde (es. un perp alt non disponibile
sull'endpoint Deribit). NON apre ordini reali: e' solo paper/lettura.
"""
from __future__ import annotations
import sys
import shutil
import tempfile
from datetime import datetime, timezone, timedelta
from pathlib import Path
import numpy as np
import pandas as pd
PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))
from src.live.cerbero_client import CerberoClient
from src.live.multi_runner import INSTRUMENT_MAP
from src.live.pairs_worker import PairsWorker
from scripts.strategies.PR01_pairs_reversion import PAIRS
def fetch(cli, asset, start, end):
inst = INSTRUMENT_MAP.get(asset, f"{asset}-PERPETUAL")
try:
cs = cli.get_historical(inst, start.strftime("%Y-%m-%d"), end.strftime("%Y-%m-%d"), "60")
if not cs:
return inst, None, "VUOTO (strumento assente sull'endpoint)"
df = pd.DataFrame(cs)
df["timestamp"] = df["timestamp"].astype("int64")
df = df.sort_values("timestamp").reset_index(drop=True)
last = pd.to_datetime(df["timestamp"].iloc[-1], unit="ms", utc=True)
age = (datetime.now(timezone.utc) - last).total_seconds() / 3600
return inst, df, f"{len(df)} barre, ultima {last:%Y-%m-%d %H:%M} ({age:.1f}h fa)"
except Exception as e:
return inst, None, f"ERRORE {type(e).__name__}: {str(e)[:60]}"
def main():
cli = CerberoClient()
end = datetime.now(timezone.utc)
start = end - timedelta(days=60)
assets = sorted({a for a, _, _ in PAIRS} | {b for _, b, _ in PAIRS})
print("=" * 80)
print(" SMOKE TEST LIVE PAIRS — fetch reale Cerbero + tick (no ordini reali)")
print("=" * 80)
data = {}
for a in assets:
inst, df, msg = fetch(cli, a, start, end)
data[a] = df
print(f" {a:<4s} [{inst:<16s}] {msg}")
print("\n tick reale per coppia:")
tmp = Path(tempfile.mkdtemp())
try:
for a, b, p in PAIRS:
if data.get(a) is None or data.get(b) is None:
print(f" {a}/{b:<4s}: SKIP (manca feed live di una gamba) -> non tradabile live ora")
continue
w = PairsWorker(a, b, "1h", params=p, fee_rt=0.001, data_dir=tmp)
w._log = lambda *x, **k: None
w._notify = lambda *x, **k: None
m = data[a][["timestamp", "close"]].merge(
data[b][["timestamp", "close"]], on="timestamp", how="inner")
if len(m) < p["n"] + 2:
print(f" {a}/{b:<4s}: merge {len(m)} barre < n+2 ({p['n']+2}) -> dati insufficienti")
continue
z, _ = w._zscore(m["close_x"].values, m["close_y"].values)
w.tick(data[a], data[b])
state = ("IN POS " + ("LONG " + a if w.direction == 1 else "SHORT " + a)
if w.in_position else "FLAT")
print(f" {a}/{b:<4s}: OK merge {len(m)} barre, z_ora={z[-1]:+.2f} -> {state}")
finally:
shutil.rmtree(tmp, ignore_errors=True)
print("\n Solo le coppie con entrambe le gambe fresche su Cerbero sono tradabili live.")
if __name__ == "__main__":
main()
+74
View File
@@ -0,0 +1,74 @@
"""Valida il PairsWorker: replay bar-per-bar sui dati storici == backtest pairs_sim?
Come validate_worker_mr01 per MR01: alimenta il PairsWorker con finestre trailing
crescenti (simula il feed live) e confronta trade/capitale finale col backtest di
riferimento scripts/analysis/pairs_research.pairs_sim. Se combaciano, la semantica
live (z-score causale, exit |z|<=z_exit o max_bars, fee 2 gambe) e' fedele.
"""
from __future__ import annotations
import shutil
import sys
import tempfile
from pathlib import Path
import pandas as pd
PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))
from src.live.pairs_worker import PairsWorker
from scripts.analysis.pairs_research import aligned, pairs_sim
from scripts.strategies.PR01_pairs_reversion import PAIRS
WINDOW = 60 # finestra trailing minima (>= n+2): z[i] corretto, replay veloce
def replay(a: str, b: str, params: dict, data_dir: Path) -> PairsWorker:
m = aligned(a, b)
df_a = m[["timestamp"]].copy(); df_a["close"] = m["close_a"].values
df_b = m[["timestamp"]].copy(); df_b["close"] = m["close_b"].values
w = PairsWorker(a, b, "1h", params=params, fee_rt=0.001, data_dir=data_dir)
# replay veloce: niente I/O su file / log / notifiche ad ogni tick (servono solo le metriche finali)
w._save_state = lambda: None
w._log = lambda *a, **k: None
w._notify = lambda *a, **k: None
n = w.n
for k in range(n + 2, len(m) + 1):
lo = max(0, k - WINDOW)
w.tick(df_a.iloc[lo:k], df_b.iloc[lo:k])
# chiudi eventuale posizione aperta a fine serie (come fa il backtest col troncamento)
return w
def main():
print("=" * 96)
print(" VALIDAZIONE PairsWorker — replay live vs backtest pairs_sim (fee 0.20% RT/coppia)")
print("=" * 96)
print(f" {'coppia':<10s}{'WORKER cap':>12s}{'trd':>5s}{'win%':>6s} | {'BACKTEST cap':>13s}{'trd':>5s}{'win%':>6s} match?")
print(" " + "-" * 88)
# Sottoinsieme rappresentativo: il codice del worker e' identico per ogni coppia,
# quindi 2 coppie con strutture diverse (alt/major e major/alt) bastano a provare
# l'equivalenza. ~135s/coppia su 73k barre orarie. Per validarle tutte: usa PAIRS.
subset = [pp for pp in PAIRS if (pp[0], pp[1]) in {("ETH", "BTC"), ("BTC", "LTC")}]
tmp = Path(tempfile.mkdtemp(prefix="pairs_validate_"))
try:
for a, b, p in subset:
w = replay(a, b, p, tmp)
bt = pairs_sim(a, b, **p)
bt_cap = 1000.0 * (1 + bt["ret"] / 100)
cap_match = abs(w.capital - bt_cap) / bt_cap < 0.02 if bt_cap else False
trd_match = abs(w.total_trades - bt["trades"]) <= max(2, bt["trades"] * 0.02)
ok = "OK" if (cap_match and trd_match) else "DIFF"
ww = w.total_wins / w.total_trades * 100 if w.total_trades else 0
print(f" {a+'/'+b:<10s}{w.capital:>12.0f}{w.total_trades:>5d}{ww:>6.1f} | "
f"{bt_cap:>13.0f}{bt['trades']:>5d}{bt['win']:>6.1f} {ok}")
finally:
shutil.rmtree(tmp, ignore_errors=True)
print(" " + "-" * 88)
print(" match = capitale entro 2% e trade entro 2% del backtest. Differenze minime sono")
print(" attese (gestione bar finale/troncamento), ma la semantica deve coincidere.")
if __name__ == "__main__":
main()
+9 -4
View File
@@ -26,10 +26,15 @@ Validazione anti-overfit (netto, fee 0.20% RT/coppia a 2 gambe, leva 3x, OOS = u
- Correlazione con BTC daily ~0.02-0.08 -> market-neutral. - Correlazione con BTC daily ~0.02-0.08 -> market-neutral.
- SCARTATA BNB/ETH: robusta solo coi suoi parametri (overfit), crolla con la universale. - SCARTATA BNB/ETH: robusta solo coi suoi parametri (overfit), crolla con la universale.
LIMITE OPERATIVO: e' una strategia a 2 gambe (long un perp + short l'altro), il worker WORKER LIVE: implementato `src/live/pairs_worker.py` (2 gambe, fee doppie, stato
attuale e' single-leg. Per tradarla serve: (a) eseguibilita' short del perp B su persistente) e wired in `multi_runner` (sezione `pairs:` in strategies.yml). Validato:
Deribit/Bybit, (b) gestione 2 ordini + fee doppie. Finche' il worker non supporta - LOGICA: `validate_worker_pairs.py` -> replay storico == backtest pairs_sim ESATTO
2 gambe, PR01 resta validata in backtest ma non wired nel paper trader. (ETH/BTC: capitale, n.trade, win% identici).
- LIVE: `live_smoke_pairs.py` (smoke reale Cerbero) -> tutte e 5 le coppie con feed
live fresco. Naming Deribit corretto: BTC/ETH = "<COIN>-PERPETUAL" (inverse),
alt = "<COIN>_USDC-PERPETUAL" (lineari USDC, storia dal 2022). Trappola: usare
"LTC-PERPETUAL"/"SOL-PERPETUAL" da' vuoto/dati sbagliati -> SEMPRE _USDC-PERPETUAL.
Resta da verificare in trading reale solo la liquidita'/fill in esecuzione.
""" """
from __future__ import annotations from __future__ import annotations
+52 -1
View File
@@ -11,6 +11,7 @@ import pandas as pd
from src.live.cerbero_client import CerberoClient from src.live.cerbero_client import CerberoClient
from src.live.strategy_loader import load_strategy from src.live.strategy_loader import load_strategy
from src.live.strategy_worker import StrategyWorker from src.live.strategy_worker import StrategyWorker
from src.live.pairs_worker import PairsWorker
from src.live.signal_engine import SignalEngine from src.live.signal_engine import SignalEngine
from src.live.telegram_notifier import send_telegram 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" DATA_DIR = PROJECT_ROOT / "data" / "paper_trades"
RESOLUTION_MAP = {"15m": "15", "1h": "60", "5m": "5"} 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 = { INSTRUMENT_MAP = {
"BTC": "BTC-PERPETUAL", "BTC": "BTC-PERPETUAL",
"ETH": "ETH-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 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(): def run():
config_path = PROJECT_ROOT / "strategies.yml" config_path = PROJECT_ROOT / "strategies.yml"
if not config_path.exists(): if not config_path.exists():
@@ -143,7 +175,8 @@ def run():
train_lookback_days = 365 train_lookback_days = 365
regular_workers, ml_workers = build_workers(config) 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: if all_worker_count == 0:
print("Nessuna strategia abilitata in strategies.yml") print("Nessuna strategia abilitata in strategies.yml")
@@ -162,6 +195,8 @@ def run():
print(f"{w.status_summary}") print(f"{w.status_summary}")
for mw in ml_workers: for mw in ml_workers:
print(f"{mw.worker.status_summary} [ML]") 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") send_telegram(f"🚀 Multi-Strategy avviato: {all_worker_count} strategie")
@@ -172,6 +207,9 @@ def run():
keys.add((w.asset, w.tf)) keys.add((w.asset, w.tf))
for mw in ml_workers: for mw in ml_workers:
keys.add((mw.worker.asset, mw.worker.tf)) 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 return keys
# Training iniziale ML # Training iniziale ML
@@ -253,6 +291,15 @@ def run():
except Exception as e: except Exception as e:
print(f" [{mw.worker.worker_id}] ERRORE: {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 # Status periodico
now = datetime.now(timezone.utc) now = datetime.now(timezone.utc)
if now.minute == 0 and now.second < poll_seconds: if now.minute == 0 and now.second < poll_seconds:
@@ -261,6 +308,8 @@ def run():
lines.append(f" {w.status_summary}") lines.append(f" {w.status_summary}")
for mw in ml_workers: for mw in ml_workers:
lines.append(f" {mw.worker.status_summary} [ML]") 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)) send_telegram("\n".join(lines))
except KeyboardInterrupt: except KeyboardInterrupt:
@@ -277,6 +326,8 @@ def run():
if df is not None and not df.empty: if df is not None and not df.empty:
mw.worker._close_position(float(df["close"].iloc[-1]), "shutdown") mw.worker._close_position(float(df["close"].iloc[-1]), "shutdown")
mw.worker._save_state() 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") send_telegram("🛑 Multi-Strategy arrestato")
break break
except Exception as e: except Exception as e:
+215
View File
@@ -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}")
+40
View File
@@ -90,3 +90,43 @@ strategies:
max_bars: 24 max_bars: 24
trend_max: 3.0 # salta fade contro trend estremo (|close-EMA200|/ATR>3): Acc+ DD- trend_max: 3.0 # salta fade contro trend estremo (|close-EMA200|/ATR>3): Acc+ DD-
ema_long: 200 ema_long: 200
# ---------------------------------------------------------------------------
# PR01 — PAIRS market-neutral spread reversion (worker a 2 GAMBE: src/live/pairs_worker.py)
# Config UNIVERSALE n50 z2 zx0.75 mb72 (anti-overfit, validata walk-forward).
# fee_rt 0.001/gamba -> 0.20% RT/coppia.
# FEED LIVE (verificato 2026-05-29): tutti i leg disponibili su Deribit via Cerbero con
# il naming corretto -> BTC/ETH = "<COIN>-PERPETUAL" (inverse), alt = "<COIN>_USDC-PERPETUAL"
# (lineari USDC, storia dal 2022). Tutte e 5 le coppie tradabili live. ETH/SOL la piu'
# debole (DD ~63%, storia SOL piu' corta) -> peso ridotto consigliato.
pairs:
- name: PR01_pairs_reversion
a: ETH
b: BTC
tf: 1h
enabled: true
params: {n: 50, z_in: 2.0, z_exit: 0.75, max_bars: 72, jump_max: 0.08}
- name: PR01_pairs_reversion
a: LTC
b: ETH
tf: 1h
enabled: true
params: {n: 50, z_in: 2.0, z_exit: 0.75, max_bars: 72, jump_max: 0.08}
- name: PR01_pairs_reversion
a: ADA
b: ETH
tf: 1h
enabled: true
params: {n: 50, z_in: 2.0, z_exit: 0.75, max_bars: 72, jump_max: 0.08}
- name: PR01_pairs_reversion
a: BTC
b: LTC
tf: 1h
enabled: true
params: {n: 50, z_in: 2.0, z_exit: 0.75, max_bars: 72, jump_max: 0.08}
- name: PR01_pairs_reversion # la piu' debole (peso ridotto)
a: ETH
b: SOL
tf: 1h
enabled: true
params: {n: 50, z_in: 2.0, z_exit: 0.75, max_bars: 72, jump_max: 0.08}