"""Verifica indipendente + ricerca PAIRS / SPREAD MEAN-REVERSION fra cripto. Famiglia nuova market-neutral (distinta da tutto l'esistente, single-asset). Idea: il log-ratio di due cripto oscilla attorno alla media; z-score estremo -> rientra. Engine ONESTO (no look-ahead, verificato): - r[i] = log(closeA[i]/closeB[i]); ma/sd = rolling(n) su r -> usano solo r[<=i]. - z[i] = (r[i]-ma[i])/sd[i]. ENTRY a close[i] (eseguibile): z<=-z_in -> LONG ratio (long A / short B); z>=+z_in -> SHORT ratio. - EXIT quando |z[j]| <= z_exit (rientro) o time-limit max_bars, a close[j]. - pairs = 2 GAMBE -> fee = 2*fee_rt*lev (0.20% RT/coppia a fee_rt=0.001), il doppio del single-asset. Rendimento neutral = retA*d - retB*d (notional uguale per gamba). - non-overlap, capitale composto. Filtro candele sporche: salta salti |dr|>jump_max. - Ritorno riportato come CAGR e Sharpe ANNUALIZZATO sul tempo reale (no sqrt(n_trade)). """ from __future__ import annotations import sys 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.data.downloader import load_data FEE_RT, LEV, POS, OOS_FRAC = 0.001, 3.0, 0.15, 0.30 BARS_YEAR = 8760 # 1h def aligned(a: str, b: str, tf: str = "1h"): da = load_data(a, tf)[["timestamp", "open", "high", "low", "close"]].rename(columns=lambda x: x + "_a" if x != "timestamp" else x) db = load_data(b, tf)[["timestamp", "close"]].rename(columns={"close": "close_b"}) m = da.merge(db, on="timestamp", how="inner").reset_index(drop=True) m["dt"] = pd.to_datetime(m["timestamp"], unit="ms", utc=True) return m def pairs_sim(a, b, tf="1h", n=50, z_in=2.0, z_exit=0.5, max_bars=72, jump_max=0.08, fee_rt=FEE_RT, lev=LEV, pos=POS, split_frac=0.0): m = aligned(a, b, tf) ca, cb = m["close_a"].values, m["close_b"].values r = np.log(ca / cb) dr = np.abs(np.diff(r, prepend=r[0])) # salto 1-bar del log-ratio ma = pd.Series(r).rolling(n).mean().values sd = pd.Series(r).rolling(n).std().values z = (r - ma) / np.where(sd == 0, np.nan, sd) # causale: usa r[<=i] ts = m["dt"]; N = len(r) split = int(N * split_frac) fee = 2 * fee_rt * lev # 2 gambe cap = peak = 1000.0; dd = 0.0; last = -1 trades = wins = 0; rets = []; yearly = {}; yearly_n = {} eq_ts: list = []; eq_v: list = [] for i in range(n + 1, N - 1): if i < split or np.isnan(z[i]) or dr[i] > jump_max: continue if i <= last: continue if z[i] <= -z_in: d = 1 elif z[i] >= z_in: d = -1 else: continue # exit: |z|<=z_exit o max_bars j = min(i + max_bars, N - 1) for k in range(1, max_bars + 1): jj = i + k if jj >= N: j = N - 1; break if abs(z[jj]) <= z_exit: j = jj; break j = jj retA = (ca[j] - ca[i]) / ca[i] retB = (cb[j] - cb[i]) / cb[i] ret = (retA - retB) * d * lev - fee # long A / short B (o viceversa) cap = max(cap + cap * pos * ret, 10.0) peak = max(peak, cap); dd = max(dd, (peak - cap) / peak) trades += 1; wins += ret > 0; rets.append(ret * pos); last = j eq_ts.append(ts.iloc[j]); eq_v.append(cap) yearly[ts.iloc[i].year] = yearly.get(ts.iloc[i].year, 0.0) + ret * 100 yearly_n[ts.iloc[i].year] = yearly_n.get(ts.iloc[i].year, 0) + 1 yrs_span = (ts.iloc[-1] - ts.iloc[max(split, 0)]).days / 365.25 or 1 sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(BARS_YEAR / np.mean([max_bars])) ) if len(rets) > 1 and np.std(rets) > 0 else 0.0 # Sharpe annualizzato sul tempo reale: usa rendimenti per-trade scalati alla frequenza media if len(rets) > 1 and np.std(rets) > 0: trades_per_year = trades / yrs_span sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(trades_per_year)) ret_tot = (cap / 1000 - 1) * 100 cagr = ((cap / 1000) ** (1 / yrs_span) - 1) * 100 if cap > 0 else -100 return dict(trades=trades, win=wins / trades * 100 if trades else 0, ret=ret_tot, cagr=cagr, dd=dd * 100, sharpe=sharpe, yearly=yearly, yearly_n=yearly_n, eq_ts=eq_ts, eq_v=eq_v) def aligned_ohlc(a: str, b: str, tf: str = "1h"): """Come aligned ma con OHLC di ENTRAMBE le gambe (serve a rilevare candele flat).""" da = load_data(a, tf)[["timestamp", "open", "high", "low", "close"]].rename( columns=lambda x: x + "_a" if x != "timestamp" else x) db = load_data(b, tf)[["timestamp", "open", "high", "low", "close"]].rename( columns=lambda x: x + "_b" if x != "timestamp" else x) m = da.merge(db, on="timestamp", how="inner").reset_index(drop=True) m["dt"] = pd.to_datetime(m["timestamp"], unit="ms", utc=True) return m def is_flat_ohlc(o, h, l, c): """Candela flat (O=H=L=C): prezzo fermo / liquidita' zero -> fill non eseguibile.""" return (o == h) & (h == l) & (l == c) def pairs_sim_flat(a, b, tf="1h", n=50, z_in=2.0, z_exit=0.5, max_bars=72, jump_max=0.08, fee_rt=FEE_RT, lev=LEV, pos=POS, split_frac=0.0, flat_skip=False, scan_buffer=192): """Engine pairs GENERALIZZATO con opzione flat-skip LIVE-REALIZABLE. Identico a pairs_sim quando flat_skip=False (regression-lock verificato). Con flat_skip=True: - ENTRY: saltata se la barra d'ingresso e' flat in UNA delle due gambe (prezzo stale). - EXIT: la condizione di uscita (|z|<=z_exit O bars>=max_bars) arma 'exit_ready'; si esce al CLOSE della PRIMA barra PULITA successiva (mai a un prezzo passato). scan_buffer = barre extra oltre max_bars concesse per trovare la barra pulita. Questa e' la stessa regola implementata nel PairsWorker live (flat_skip) -> parita'. """ m = aligned_ohlc(a, b, tf) ca, cb = m["close_a"].values, m["close_b"].values N = len(ca) if flat_skip: flat = (is_flat_ohlc(m["open_a"].values, m["high_a"].values, m["low_a"].values, ca) | is_flat_ohlc(m["open_b"].values, m["high_b"].values, m["low_b"].values, cb)) else: flat = np.zeros(N, dtype=bool) r = np.log(ca / cb) dr = np.abs(np.diff(r, prepend=r[0])) ma = pd.Series(r).rolling(n).mean().values sd = pd.Series(r).rolling(n).std().values z = (r - ma) / np.where(sd == 0, np.nan, sd) ts = m["dt"] split = int(N * split_frac) fee = 2 * fee_rt * lev cap = peak = 1000.0; dd = 0.0; last = -1 trades = wins = 0; rets = []; yearly = {}; yearly_n = {} eq_ts, eq_v = [], [] n_skip_entry = 0 kmax = max_bars + (scan_buffer if flat_skip else 0) for i in range(n + 1, N - 1): if i < split or np.isnan(z[i]) or dr[i] > jump_max or i <= last: continue if z[i] <= -z_in: d = 1 elif z[i] >= z_in: d = -1 else: continue if flat[i]: n_skip_entry += 1 continue # niente ingresso su barra stale # uscita live-realizable: arma a |z|<=z_exit o max_bars, esci alla prima barra pulita exit_ready = False; j = i for k in range(1, kmax + 1): jj = i + k if jj >= N: j = N - 1; break if not exit_ready and (abs(z[jj]) <= z_exit or k >= max_bars): exit_ready = True if exit_ready and not flat[jj]: j = jj; break j = jj retA = (ca[j] - ca[i]) / ca[i] retB = (cb[j] - cb[i]) / cb[i] ret = (retA - retB) * d * lev - fee cap = max(cap + cap * pos * ret, 10.0) peak = max(peak, cap); dd = max(dd, (peak - cap) / peak) trades += 1; wins += ret > 0; rets.append(ret * pos); last = j eq_ts.append(ts.iloc[j]); eq_v.append(cap) yearly[ts.iloc[i].year] = yearly.get(ts.iloc[i].year, 0.0) + ret * 100 yearly_n[ts.iloc[i].year] = yearly_n.get(ts.iloc[i].year, 0) + 1 yrs_span = (ts.iloc[-1] - ts.iloc[max(split, 0)]).days / 365.25 or 1 sharpe = 0.0 if len(rets) > 1 and np.std(rets) > 0: sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(trades / yrs_span)) ret_tot = (cap / 1000 - 1) * 100 cagr = ((cap / 1000) ** (1 / yrs_span) - 1) * 100 if cap > 0 else -100 return dict(trades=trades, win=wins / trades * 100 if trades else 0, ret=ret_tot, cagr=cagr, dd=dd * 100, sharpe=sharpe, yearly=yearly, yearly_n=yearly_n, eq_ts=eq_ts, eq_v=eq_v, n_skip_entry=n_skip_entry) def check_no_lookahead(): """Perturba il FUTURO del ratio e verifica che z[i] non cambi (causalita').""" m = aligned("ETH", "BTC") r = np.log(m["close_a"].values / m["close_b"].values) n = 50; i = 1000 z_i = (r[i] - pd.Series(r).rolling(n).mean().values[i]) / pd.Series(r).rolling(n).std().values[i] r2 = r.copy(); r2[i + 1:] += 0.5 # stravolge il futuro z_i2 = (r2[i] - pd.Series(r2).rolling(n).mean().values[i]) / pd.Series(r2).rolling(n).std().values[i] print(f" no-look-ahead: z[i]={z_i:.6f} vs z[i] con futuro perturbato={z_i2:.6f} -> " f"{'OK (identico)' if abs(z_i - z_i2) < 1e-9 else 'VIOLAZIONE!'}") def main(): print("=" * 104) print(f" PAIRS spread reversion — NETTO fee 0.20% RT/coppia (2 gambe), leva {LEV:.0f}x, OOS ultimo {int(OOS_FRAC*100)}%") print("=" * 104) check_no_lookahead() pairs = [("ETH", "BTC"), ("LTC", "ETH"), ("ADA", "ETH"), ("SOL", "ETH"), ("BNB", "BTC"), ("XRP", "BTC"), ("SOL", "BTC"), ("DOGE", "BTC")] print(f"\n {'coppia':<10s}{'trd':>5s}{'win%':>6s}{'FULL%':>8s}{'OOS%':>8s}{'CAGR%':>7s}" f"{'DD%':>6s}{'oDD%':>6s}{'Shrp':>6s}{'anni+':>7s}{'fee0.4%RT':>11s}") print(" " + "-" * 96) for a, b in pairs: f = pairs_sim(a, b, n=50, z_in=2.0, z_exit=0.5, max_bars=72) o = pairs_sim(a, b, n=50, z_in=2.0, z_exit=0.5, max_bars=72, split_frac=1 - OOS_FRAC) hi = pairs_sim(a, b, n=50, z_in=2.0, z_exit=0.5, max_bars=72, fee_rt=0.002) # 0.4% RT/coppia yrs = f["yearly"]; pos_y = sum(1 for v in yrs.values() if v > 0) print(f" {a+'/'+b:<10s}{f['trades']:>5d}{f['win']:>6.1f}{f['ret']:>+8.0f}{o['ret']:>+8.0f}" f"{f['cagr']:>7.0f}{f['dd']:>6.0f}{o['dd']:>6.0f}{f['sharpe']:>6.2f}{f'{pos_y}/{len(yrs)}':>7s}" f"{hi['ret']:>+11.0f}") # correlazione con BTC daily (market-neutrality) sulla coppia migliore print("\n Verifica market-neutrality ETH/BTC: per-anno") f = pairs_sim("ETH", "BTC", n=50, z_in=2.0, z_exit=0.5, max_bars=72) print(" " + " ".join(f"{y}:{v:+.0f}%" for y, v in sorted(f["yearly"].items()))) if __name__ == "__main__": main()