"""r0724_onchain_wave — prima ondata ON-CHAIN / SENTIMENT (famiglia MAI testata) (2026-07-24). Goal "trova altre strategie": la famiglia on-chain e' l'unica grande famiglia di INFORMAZIONE (non di prezzo) mai toccata dal progetto. Fonte segnali: CoinMetrics Community (github.com/coinmetrics/data, CSV daily dal genesis, gratuito) + Fear&Greed (alternative.me, dal 2018-02). I RITORNI restano SOLO dal feed certificato Deribit (lezione v2.0.0: i dati esterni sono segnale, mai prezzo). 5 famiglie, ognuna giudicata con `study_family_honest` (selezione cella IN-SAMPLE, deflated-Sharpe sull'INTERA griglia, marginal scorer indurito vs TP01): NET — network-growth momentum (AdrActCnt / TxTfrCnt, stile Liu-Tsyvinski) MVRV — valuation gate su CapMVRVCur (percentile causale espandente) EXS — supply su exchange (SplyExNtv: accumulo = coin che LASCIANO gli exchange) HASH — hash ribbons BTC-only (capitolazione/recovery miner; ETH post-Merge = 0) FNG — Fear&Greed contrarian (long dopo paura estrema) CAUSALITA' (doppio lag): la riga CoinMetrics del giorno d si completa a fine giorno d (+ ore di processing) -> al close del bar daily d (=00:00 UTC di d+1) l'ultima riga SICURAMENTE nota e' d-1 => segnale shiftato di 1 GIORNO prima del mapping sui bar; eval_weights shifta di un altro bar (decisione a close[i], hold i+1) => lag totale attivita'->posizione = 2 giorni. F&G: pubblicato ~00:00 UTC del giorno stesso -> stesso trattamento conservativo. CAVEAT DATI (dichiarati): CM community aggiornato al 2026-05-24 (~2 mesi di lag: ok per ricerca, NON per un deploy senza fonte fresca); metriche exchange-flow = stima CM dei wallet exchange (proxy, non verita'); F&G e' in parte DERIVATO dal prezzo (vol+momentum) -> rischio ridondanza col trend, il marginal scorer lo vede. Uso: `uv run python scripts/research/r0724_onchain_wave.py` Dati attesi in data/external/coinmetrics/ (cm_btc.csv, cm_eth.csv, fng.json); per aggiornare: scaricare di nuovo dalle fonti (URL nei commenti di _load_cm/_load_fng). """ from __future__ import annotations import json import sys from pathlib import Path import numpy as np import pandas as pd ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt")) import altlib # noqa: E402 from altlib import study_family_honest, fmt_marginal # noqa: E402 EXT = ROOT / "data" / "external" / "coinmetrics" AVAIL_LAG_D = 1 # giorni di lag di disponibilita' del segnale PRIMA del mapping sui bar # ------------------------------------------------------------------ dati segnale def _load_cm(asset: str) -> pd.DataFrame: """CoinMetrics community: https://raw.githubusercontent.com/coinmetrics/data/master/csv/{btc,eth}.csv""" df = pd.read_csv(EXT / f"cm_{asset.lower()}.csv", low_memory=False) idx = pd.DatetimeIndex(pd.to_datetime(df["time"], utc=True)).floor("D") out = df.drop(columns=["time"]).apply(pd.to_numeric, errors="coerce") out.index = idx return out def _load_fng() -> pd.Series: """alternative.me: https://api.alternative.me/fng/?limit=0&format=json""" d = json.loads((EXT / "fng.json").read_text())["data"] ts = pd.to_datetime([int(x["timestamp"]) for x in d], unit="s", utc=True).floor("D") return pd.Series([float(x["value"]) for x in d], index=ts).sort_index() def _load_stables() -> pd.Series: """DefiLlama: https://stablecoins.llama.fi/stablecoincharts/all -> supply USD totale stablecoin per giorno (dal 2017-11). Liquidita' 'dry powder' NON derivata dal prezzo BTC.""" d = json.loads((EXT / "stables.json").read_text()) ts = pd.to_datetime([int(x["date"]) for x in d], unit="s", utc=True).floor("D") v = [float(x.get("totalCirculating", {}).get("peggedUSD", np.nan)) for x in d] return pd.Series(v, index=ts).sort_index() _CM = {a: _load_cm(a) for a in ("BTC", "ETH")} _FNG = _load_fng() _STB = _load_stables() def _to_target(df: pd.DataFrame, sig_by_day: pd.Series) -> np.ndarray: """Mappa un segnale by-day sui bar del df certificato con lag di disponibilita'. tz-aware su entrambi i lati (lezione della sera: naive-vs-aware nel reindex = NaN->0 silenziosi).""" days = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)).floor("D") known = sig_by_day.shift(AVAIL_LAG_D) return np.nan_to_num(known.reindex(days).values.astype(float)) def _expanding_pctl(x: pd.Series, minp: int = 365) -> pd.Series: """Percentile causale espandente di x[t] nella storia fino a t (incluso).""" v = x.values.astype(float) out = np.full(len(v), np.nan) order: list[float] = [] import bisect for i, xi in enumerate(v): if np.isfinite(xi): bisect.insort(order, xi) if len(order) >= minp: out[i] = bisect.bisect_left(order, xi) / len(order) return pd.Series(out, index=x.index) # ------------------------------------------------------------------ famiglie def net_factory(tf: str, metric: str = "AdrActCnt", L: int = 30, mode: str = "LF"): def fn(df, asset): m = _CM[asset][metric] sm = np.log(m.rolling(7, min_periods=4).mean()) sig = sm - sm.shift(L) pos = np.sign(sig) if mode == "LS" else (sig > 0).astype(float) return _to_target(df, pos) return fn def mvrv_factory(tf: str, lo: float = 0.3, hi: float = 1.1, mode: str = "LF"): """LF lo: long solo se MVRV-pctl < lo (compra paura di valuation). LF hi: long salvo top (pctl < hi). LS: +1 sotto lo, -1 sopra hi.""" def fn(df, asset): p = _expanding_pctl(_CM[asset]["CapMVRVCur"]) if mode == "LS": pos = pd.Series(np.where(p < lo, 1.0, np.where(p > hi, -1.0, 0.0)), index=p.index) else: pos = (p < (lo if hi > 1.0 else hi)).astype(float) if hi > 1.0 else (p < hi).astype(float) return _to_target(df, pos) return fn def exs_factory(tf: str, L: int = 30, mode: str = "LF"): def fn(df, asset): s = np.log(_CM[asset]["SplyExNtv"].where(_CM[asset]["SplyExNtv"] > 0)) sig = -(s - s.shift(L)) # supply che LASCIA gli exchange = accumulo = + pos = np.sign(sig) if mode == "LS" else (sig > 0).astype(float) return _to_target(df, pos) return fn def hash_factory(tf: str, fast: int = 30, slow: int = 60, mode: str = "ribbon"): """BTC-only. ribbon: long da recovery-cross (fast riattraversa sopra slow) fino alla prossima capitolazione (fast sotto slow). holdH: long per H giorni dal recovery.""" def fn(df, asset): if asset != "BTC": return np.zeros(len(df)) h = _CM[asset]["HashRate"] f, s = h.rolling(fast).mean(), h.rolling(slow).mean() above = (f > s).astype(float) if mode == "ribbon": pos = above # long quando il ribbon e' sano else: # recovery: long H giorni dal cross-up H = int(mode[4:]) cross_up = (above.diff() > 0) pos = cross_up.rolling(H, min_periods=1).max().fillna(0.0) return _to_target(df, pos) return fn def fng_factory(tf: str, lo: int = 20, H: int = 30, mode: str = "fear"): """fear: long H giorni dopo F&G < lo (contrarian). regime: long quando media7 > 50.""" def fn(df, asset): g = _FNG if mode == "regime": pos = (g.rolling(7, min_periods=4).mean() > 50).astype(float) else: trig = (g < lo) pos = trig.rolling(H, min_periods=1).max().fillna(0.0) return _to_target(df, pos) return fn def stable_factory(tf: str, L: int = 30, mode: str = "LF", thr: float = 0.0): """Crescita della supply stablecoin totale (liquidita' in ingresso nel sistema). LF: long se crescita L-giorni > thr (annualizzata), flat altrimenti. LS: segno.""" def fn(df, asset): s = np.log(_STB.where(_STB > 0)) sig = (s - s.shift(L)) * (365.0 / L) - thr pos = np.sign(sig) if mode == "LS" else (sig > 0).astype(float) return _to_target(df, pos) return fn FAMILIES = [ ("STABLE-supply-growth", stable_factory, [ dict(L=30, mode="LF", thr=0.0), dict(L=90, mode="LF", thr=0.0), dict(L=30, mode="LF", thr=0.10), dict(L=90, mode="LF", thr=0.10), dict(L=30, mode="LS", thr=0.0), dict(L=90, mode="LS", thr=0.0), ]), ("NET-growth", net_factory, [ dict(metric="AdrActCnt", L=30, mode="LF"), dict(metric="AdrActCnt", L=90, mode="LF"), dict(metric="AdrActCnt", L=30, mode="LS"), dict(metric="AdrActCnt", L=90, mode="LS"), dict(metric="TxTfrCnt", L=30, mode="LF"), dict(metric="TxTfrCnt", L=90, mode="LF"), dict(metric="TxTfrCnt", L=30, mode="LS"), dict(metric="TxTfrCnt", L=90, mode="LS"), ]), ("MVRV-valuation", mvrv_factory, [ dict(lo=0.2, hi=9.9, mode="LF"), dict(lo=0.3, hi=9.9, mode="LF"), dict(lo=0.0, hi=0.8, mode="LF"), dict(lo=0.0, hi=0.9, mode="LF"), dict(lo=0.2, hi=0.8, mode="LS"), dict(lo=0.3, hi=0.9, mode="LS"), ]), ("EXS-exchange-supply", exs_factory, [ dict(L=30, mode="LF"), dict(L=90, mode="LF"), dict(L=30, mode="LS"), dict(L=90, mode="LS"), ]), ("HASH-ribbons-BTC", hash_factory, [ dict(fast=30, slow=60, mode="ribbon"), dict(fast=30, slow=60, mode="hold60"), dict(fast=30, slow=60, mode="hold120"), ]), ("FNG-fear-greed", fng_factory, [ dict(lo=15, H=10, mode="fear"), dict(lo=15, H=30, mode="fear"), dict(lo=25, H=10, mode="fear"), dict(lo=25, H=30, mode="fear"), dict(mode="regime"), ]), ] def main() -> None: print("=" * 100) print(" ONDATA ON-CHAIN / SENTIMENT — 5 famiglie via study_family_honest (gate completi)") print(f" CM: BTC {_CM['BTC'].index[0].date()}->{_CM['BTC'].index[-1].date()}, " f"ETH {_CM['ETH'].index[0].date()}->{_CM['ETH'].index[-1].date()} | " f"F&G {_FNG.index[0].date()}->{_FNG.index[-1].date()}") print("=" * 100) n_cells_tot = sum(len(g) for _, _, g in FAMILIES) print(f" trial totali dichiarati: {n_cells_tot} celle su 5 famiglie " "(tutte contate nel deflated-Sharpe di famiglia)\n") results = [] for name, factory, grid in FAMILIES: print("-" * 100) rep = study_family_honest(name, factory, grid, tfs=("1d",)) results.append(rep) if rep.get("chosen") is None: print(f"=== {name}: nessuna cella valida in-sample") continue ch = rep["chosen"] print(f"=== {name}: cella IS {ch['params']} (IS Sh {ch['insample_sharpe']}, " f"full {ch['full_sharpe']}) su {rep['n_cells']} celle") print(f" deflated-Sharpe {rep['deflated_sharpe']} (null-max atteso " f"{rep['expected_null_max']}) dsr_pass={rep['dsr_pass']}") print(fmt_marginal(rep["marginal"]) if isinstance(rep["marginal"], str) else fmt_marginal(rep["marginal"])) print(f" >>> EARNS_SLOT_HONEST = {rep['earns_slot_honest']}") print("\n" + "=" * 100) print(" SINTESI") print("=" * 100) for rep in results: ch = rep.get("chosen") lab = "no-cell" if ch is None else ( f"IS{ch['params']} dsr={rep.get('deflated_sharpe')} " f"marg={rep['marginal'].get('marginal_verdict') if isinstance(rep.get('marginal'), dict) else rep['marginal']['marginal_verdict']}") print(f" {rep['name']:<24} earns_slot_honest={rep.get('earns_slot_honest')} {lab}") if __name__ == "__main__": main()