PythagorasGoal/scripts/research/r0724_onchain_wave.py

"""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()