Files

1213 lines
64 KiB
Python

#!/usr/bin/env python
"""r0822d_funding.py — IL FUNDING DEI PERPETUAL: il costo che il progetto non ha mai prezzato.
CONTESTO (ondata 2026-08-22, sezione 33 di docs/research/RESULTS-0822.md). Verificato dal
coordinatore: **zero occorrenze della parola `funding`** in `src/backtest/harness.py`,
`src/strategies/trend_portfolio.py`, `src/portfolio/{sleeves,portfolio}.py` e in tutto
`src/live/`. Il progetto ha prezzato fee, slippage, min-order, pavimento IB, haircut small-cap,
fortuna d'ancora, degrado d'esecuzione, look-ahead, split, contaminazione EUR/USD, rischio di
venue e fisco — e MAI il costo di TENERE APERTA una posizione.
La stima grezza che ha fatto emergere il buco: proxy Hyperliquid 14,8%/anno x esposizione long
media di TP01 0,103·E = ~1,53%/anno = il 9% del drift de-luckato. Questo script la sostituisce
con una MISURA.
COSA FA (un grado di liberta' per volta, replica di controllo prima di ogni delta):
0. IL DATO. Storico funding da `public/get_funding_rate_history` di Deribit per i QUATTRO
perpetual: i LINEARI USDC (`BTC_USDC-PERPETUAL`/`ETH_USDC-PERPETUAL`, che sono cio' che il
libro esegue davvero) e gli INVERSE (`BTC-PERPETUAL`/`ETH-PERPETUAL`, che hanno la storia
lunga). Copertura dichiarata, buchi contati, accordo linear-vs-inverse MISURATO sulla
finestra comune. Cache su scratchpad; pacing <=2 req/s; mai nei minuti :07 (cron_book) e
:25-:29 (cron_chain).
1. REPLICA DI CONTROLLO. Con funding SPENTO le repliche devono riprodurre BIT-EXACT
`al.tp01_baseline_daily`, `CC.tp01_realistic`, `r02.sim_equity` e `CC.book_series('hourly')`.
Nessun delta viene stampato prima che questa sezione passi.
2. IL SEGNO. Il funding si PAGA da long in contango e si INCASSA da short. TP01 e' long-flat
(paga quasi sempre), SKH01 e' long/short e sta a mercato una frazione del tempo (puo'
incassare). Il segno NON e' assunto: e' misurato appaiando la posizione oraria/giornaliera
di ciascuno sleeve col funding del SUO strumento.
3. I NUMERI. Sharpe FULL/IS/HOLD, maxDD, CAGR, drift di TP01, SKH01 e del libro 75/25, con e
senza funding, e il DELTA su ciascuno.
4. LE DECISIONI. Soffitto ~1,3 · peso 75/25 (sweep w x banda d'ancora, mediana delle
DIFFERENZE APPAIATE) · muri di capitale e rendita perpetua (che vivono sul DRIFT) ·
k* e il gradino di leva (il funding e' proporzionale all'esposizione, quindi scala con k).
5. CONTROLLO SUL CONTO VERO. Il libro esegue dal 2026-07-08: posizione oraria ricostruita da
`logs/cron_book.log`, funding modellato coi tassi REALI, confronto col `realized_pnl` che
il conto riporta sulle due posizioni aperte.
6. COSA NON E' CATTURATO, quantificato: l'approssimazione "esposizione media x tasso medio"
contro l'accrual esatto per trade (su SKH01, che entra ed esce), il proxy inverse prima
del 2022-03, la discretezza del pagamento.
VINCOLI: nessun file di produzione toccato (config/, src/, scripts/live/, cron INVARIATI);
nessun commit; output solo a video + cache in scratchpad. Runtime ~2-6 min con cache calda.
Uso: `uv run python scripts/research/r0822d_funding.py`
`--no-net` usa solo la cache (nessuna richiesta a Deribit)
"""
from __future__ import annotations
import datetime as dt
import json
import re
import sys
import time
import urllib.request
from pathlib import Path
import numpy as np
import pandas as pd
ROOT = Path("/opt/docker/PythagorasGoal")
sys.path.insert(0, str(ROOT / "scripts" / "research"))
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))
sys.path.insert(0, str(ROOT))
import altlib as al # noqa: E402
import r0702_anchor_skh01 as r02 # noqa: E402 (run_asset / sim_equity / get5m)
import r0724_skh_live_weight as skl # noqa: E402 (skh_book, W_GRID, OFFSETS)
import r0725_capcurve as CC # noqa: E402 (tp01_realistic, book_series, survival)
import r0726_capwall_refresh as WR # noqa: E402 (perp_and_wall, trajectory)
from src.data.downloader import load_data # noqa: E402
from src.portfolio.portfolio import combine_outer, metrics # noqa: E402
from src.strategies.trend_portfolio import ( # noqa: E402
CANONICAL, TrendPortfolio, resample_1d, simple_returns)
# ---------------------------------------------------------------------------
# Costanti dichiarate
# ---------------------------------------------------------------------------
# Cache dello storico funding FUORI dal repo (`data/raw/` non e' toccabile da questo filone,
# e questi parquet non sono dato certificato: sono una cache ri-scaricabile in ~4 minuti).
_SP = Path("/tmp/claude-1001/-opt-docker-PythagorasGoal/"
"b6cc75e7-14f8-4c32-bd07-ab8a0d2aaee6/scratchpad/fund")
SCRATCH = _SP if _SP.parent.exists() else Path("/tmp/pythagoras_funding_cache")
API = "https://www.deribit.com/api/v2/public/get_funding_rate_history"
SPAN_MS = 30 * 24 * 3600 * 1000 # 30 giorni per richiesta (l'endpoint da' 744 record)
HOLDOUT = al.HOLDOUT
ASSETS = ("BTC", "ETH")
MSH = 3_600_000
ALLOC = 600.0 # il capitale su cui gira book_series (conto reale $635)
DELUCK = 0.89 # fattore d'ancora MISURATO (r0726_deluck_factor)
EURUSD = CC.EURUSD
N_PATHS_WALL = 2000 # risoluzione MC dichiarata (il muro e' +-1%)
# strumento -> (data di creazione su Deribit, asset)
INSTR = {
"BTC_USDC-PERPETUAL": ("2022-03-09", "BTC", "linear"),
"ETH_USDC-PERPETUAL": ("2022-03-14", "ETH", "linear"),
"BTC-PERPETUAL": ("2018-08-14", "BTC", "inverse"),
"ETH-PERPETUAL": ("2019-03-14", "ETH", "inverse"),
}
LINEAR = {"BTC": "BTC_USDC-PERPETUAL", "ETH": "ETH_USDC-PERPETUAL"}
INVERSE = {"BTC": "BTC-PERPETUAL", "ETH": "ETH-PERPETUAL"}
USE_NET = "--no-net" not in sys.argv
T0 = time.time()
def sez(t: str) -> None:
print("\n" + "=" * 100)
print(f" {t}")
print("=" * 100)
def clock() -> str:
return f"[{time.time() - T0:5.0f}s]"
# ===========================================================================
# 0) IL DATO — storico funding Deribit
# ===========================================================================
def _pace() -> None:
"""<=2 req/s, e mai nei minuti :07 (cron_book) e :25-:29 (cron_chain)."""
while dt.datetime.now(dt.timezone.utc).minute in (7, 25, 26, 27, 28, 29):
time.sleep(20)
time.sleep(0.55)
def fetch_funding(instr: str, start_iso: str) -> pd.DataFrame:
"""Storico orario del funding, incrementale su cache parquet."""
SCRATCH.mkdir(parents=True, exist_ok=True)
out = SCRATCH / f"{instr.replace('-', '_')}.parquet"
have = pd.read_parquet(out) if out.exists() else pd.DataFrame()
if not USE_NET:
if not len(have):
raise RuntimeError(f"--no-net ma cache assente per {instr}")
return have
t0 = int(pd.Timestamp(start_iso, tz="UTC").timestamp() * 1000)
if len(have):
t0 = int(have["ts"].max()) - MSH
tend = int(pd.Timestamp.now(tz="UTC").floor("h").timestamp() * 1000)
rows, a = [], t0
while a < tend:
b = min(a + SPAN_MS, tend)
u = f"{API}?instrument_name={instr}&start_timestamp={a}&end_timestamp={b}"
for att in range(4):
try:
_pace()
r = json.load(urllib.request.urlopen(u, timeout=30))["result"]
break
except Exception:
if att == 3:
raise
time.sleep(3 * (att + 1))
rows.extend({"ts": x["timestamp"], "i1h": x["interest_1h"],
"i8h": x["interest_8h"], "index": x["index_price"]} for x in r)
a = b
df = pd.concat([have, pd.DataFrame(rows)], ignore_index=True) if len(have) else pd.DataFrame(rows)
df = df.drop_duplicates(subset="ts").sort_values("ts").reset_index(drop=True)
df.to_parquet(out, index=False)
return df
def sezione0() -> dict:
sez("0) IL DATO — storico funding Deribit (endpoint pubblico get_funding_rate_history)")
print(" Il libro esegue sui LINEARI USDC. Gli INVERSE sono qui solo perche' hanno la storia")
print(" lunga: servono come PROXY prima del 2022-03, e l'accordo fra i due si MISURA.")
print(" (4a e 5a occorrenza dello schema `fee_watch`: puntare la misura sull'inverse")
print(" avrebbe misurato uno strumento che il libro non trada.)\n")
F = {}
print(f" {'strumento':<22} {'righe':>7} {'da':>12} {'a':>12} {'ore attese':>11} "
f"{'buchi':>6} {'ann. medio':>11}")
for instr, (start, asset, kind) in INSTR.items():
d = fetch_funding(instr, start)
d = d.copy()
d["dt"] = pd.to_datetime(d["ts"], unit="ms", utc=True)
d = d.drop_duplicates(subset="ts").sort_values("dt").set_index("dt")
span_h = int((d.index.max() - d.index.min()) / pd.Timedelta("1h")) + 1
gaps = span_h - len(d)
ann = float(d["i1h"].mean()) * 24 * 365.25
F[instr] = d
print(f" {instr:<22} {len(d):>7,} {str(d.index.min().date()):>12} "
f"{str(d.index.max().date()):>12} {span_h:>11,} {gaps:>6,} {ann:>10.2%}")
b0 = pd.Timestamp("2019-03-14", tz="UTC")
sw = min(F[LINEAR[a]].index.min() for a in ASSETS)
tot_d = (F[LINEAR["BTC"]].index.max() - b0).days
print("\n ⚠️ VINCOLO DEL FILONE, da dire subito: i perpetual LINEARI USDC esistono dal")
print(f" {sw.date()}. Il backtest del libro parte dal 2019-03-14 -> per "
f"{(sw-b0).days/tot_d:.0%} del campione ({(sw-b0).days:,} giorni su {tot_d:,}) il")
print(" funding dello strumento ESEGUITO non esiste. Si usa l'inverse come proxy e si")
print(" misura quanto i due differiscono dove coesistono.")
print("\n ANNO per ANNO (annualizzato, ore x 24 x 365.25):")
print(f" {'anno':>6} " + " ".join(f"{k.replace('-PERPETUAL',''):>12}" for k in INSTR))
yrs = sorted({y for d in F.values() for y in d.index.year.unique()})
for y in yrs:
cells = []
for k in INSTR:
d = F[k]
g = d[d.index.year == y]["i1h"]
cells.append(f"{g.mean()*24*365.25:>11.2%}" if len(g) > 100 else f"{'—':>12}")
print(f" {y:>6} " + " ".join(cells))
print("\n CONVENZIONE TEMPORALE — verificata dai dati, non assunta dalla documentazione.")
print(" Se il record etichettato T copre l'ora CHE FINISCE in T, allora `interest_8h` deve")
print(" essere la somma degli 8 `interest_1h` fino a T INCLUSO. Le due alternative danno")
print(" numeri diversi e si vedono:")
print(f" {'ipotesi':<26} {'corr':>9} {'mediana |dif|':>15}")
d = F["BTC_USDC-PERPETUAL"]
h = {"trailing, T incluso": d["i1h"].rolling(8).sum(),
"trailing, T escluso": d["i1h"].shift(1).rolling(8).sum(),
"leading (T -> T+8h)": d["i1h"][::-1].rolling(8).sum()[::-1].shift(-1)}
for lbl, x in h.items():
m = (~x.isna()) & (~d["i8h"].isna())
print(f" {lbl:<26} {float(x[m].corr(d['i8h'][m])):>9.6f} "
f"{float((x[m]-d['i8h'][m]).abs().median()):>15.3e}")
print(" -> 'trailing, T incluso'. E' la convenzione usata da tutto lo script.")
# accordo linear vs inverse sulla finestra comune
print("\n ACCORDO LINEAR vs INVERSE sulla finestra comune (ore appaiate):")
print(f" {'asset':>6} {'ore':>8} {'ann.LIN':>9} {'ann.INV':>9} {'corr':>7} "
f"{'diff annua':>11} {'|diff| med/ora':>15}")
agree = {}
for a in ASSETS:
L, I = F[LINEAR[a]]["i1h"], F[INVERSE[a]]["i1h"]
J = pd.concat({"L": L, "I": I}, axis=1, join="inner").dropna()
d_ann = float((J["L"] - J["I"]).mean()) * 24 * 365.25
agree[a] = dict(n=len(J), lin=float(J["L"].mean()) * 24 * 365.25,
inv=float(J["I"].mean()) * 24 * 365.25,
corr=float(J["L"].corr(J["I"])), diff=d_ann)
print(f" {a:>6} {len(J):>8,} {agree[a]['lin']:>8.2%} {agree[a]['inv']:>8.2%} "
f"{agree[a]['corr']:>7.3f} {d_ann:>10.2%} {float((J['L']-J['I']).abs().median()):>15.2e}")
print("\n IL PROXY DA CUI E' NATO IL FILONE — Hyperliquid (data/raw/hlfund_*_1h.parquet)")
print(" contro il venue su cui il libro esegue davvero, sulle ore APPAIATE:")
print(f" {'asset':>6} {'ore comuni':>11} {'ann. HL':>9} {'ann. Deribit':>13} {'rapporto':>9}")
for a in ASSETS:
try:
hl = pd.read_parquet(ROOT / "data" / "raw" / f"hlfund_{a.lower()}_1h.parquet")["funding"]
except Exception:
print(f" {a:>6} {'—':>11} (file assente)")
continue
hl.index = pd.DatetimeIndex(hl.index).floor("h")
hl = hl[~hl.index.duplicated(keep="last")]
de = F[LINEAR[a]]["i1h"]
J = pd.concat({"hl": hl, "de": de}, axis=1, join="inner").dropna()
print(f" {a:>6} {len(J):>11,} {float(J['hl'].mean())*24*365.25:>9.2%} "
f"{float(J['de'].mean())*24*365.25:>13.2%} "
f"{float(J['hl'].mean())/float(J['de'].mean()):>8.2f}x")
print(" ⚠️ Il funding NON e' una proprieta' del mercato, e' una proprieta' del VENUE (formula")
print(" e cap diversi). Usare il tasso di Hyperliquid per stimare il costo di un libro che")
print(" esegue su Deribit e' lo stesso errore di `fee_watch` il 21/08 — misurare uno")
print(" strumento che non si trada — con l'aggravante che qui non e' nemmeno lo stesso")
print(" exchange. (Il numero finale esce comunque piu' ALTO della stima grezza: due")
print(" errori in direzioni opposte, il tasso troppo alto e l'esposizione condizionale")
print(" troppo bassa. Compensarsi non e' misurare.)")
print(" 📌 E ha una CONSEGUENZA di venue: se a $20k si riapre lo split e XS01 va su HL, il")
print(" suo funding va prezzato col tasso di HL, non con questo — ed e' un altro numero.")
return dict(F=F, agree=agree)
def build_rate(F: dict) -> dict:
"""Serie oraria del funding per asset: LINEARE dove esiste, INVERSE prima (proxy dichiarato).
Ritorna, per asset: ts (ms, fine dell'ora), i1h, cumF (somma cumulata), e la data di switch."""
out = {}
for a in ASSETS:
lin, inv = F[LINEAR[a]]["i1h"], F[INVERSE[a]]["i1h"]
sw = lin.index.min()
merged = pd.concat([inv[inv.index < sw], lin]).sort_index()
merged = merged[~merged.index.duplicated(keep="last")]
ts = to_ms(merged.index)
v = merged.values.astype(float)
out[a] = dict(ts=ts, i1h=v, cum=np.concatenate([[0.0], np.cumsum(v)]),
switch=sw, series=merged)
return out
EPOCH = pd.Timestamp("1970-01-01", tz="UTC")
def to_ms(idx) -> np.ndarray:
"""Epoca in MILLISECONDI, esplicita.
⚠️ NON usare `.astype("int64")` / `.view("int64")` su un DatetimeIndex: in pandas 2.x la
risoluzione non e' piu' sempre ns e la scala esce SBAGLIATA in modo silenzioso. Misurato
qui, in questa sessione, su tre serie dello stesso progetto: il funding cachato ha unita'
`datetime64[ms]`, l'indice di `al.get()` ha `datetime64[us]`, quello del log del cron ha
`datetime64[s]`. Con `.astype("int64")//1e6` le prime due finiscono su scale che
differiscono di 1.000x -> `searchsorted` restituisce l'ESTREMO dell'array invece
dell'ora giusta, e il funding usciva 0.000%/anno su SKH01 (troppo piccolo) e
+1.229%/GIORNO su TP01 (troppo grande). Nessuna eccezione, nessun NaN: due numeri
plausibili in due punti diversi, e uno dei due sarebbe passato per buono.
(Stessa famiglia dell'errore gia' codificato in CLAUDE.md il 2026-07-01.)
"""
if not isinstance(idx, pd.DatetimeIndex):
idx = pd.DatetimeIndex(idx)
if idx.tz is None:
idx = idx.tz_localize("UTC")
return ((idx - EPOCH) // pd.Timedelta(milliseconds=1)).values.astype(np.int64)
def cumF(rate: dict, t_ms) -> np.ndarray:
"""Funding cumulato fino all'istante t (record etichettato T = ora CHIUSA in (T-1h, T])."""
j = np.searchsorted(rate["ts"], np.asarray(t_ms, dtype=np.int64), side="right")
return rate["cum"][j]
def daily_funding(rate: dict, days: pd.DatetimeIndex) -> np.ndarray:
"""Funding pagato da un long unitario nel giorno d = somma di i1h su (d, d+24h]."""
t0 = to_ms(days)
return cumF(rate, t0 + 24 * MSH) - cumF(rate, t0)
def verifica_scala(rate: dict, F: dict) -> None:
"""Controllo POSITIVO sulla scala dei tempi (senza, il bug qui sotto ripassa muto).
Non basta che i numeri sembrino plausibili: `daily_funding` (searchsorted su epoca in ms)
si confronta con un `groupby` di pandas, che non fa nessuna conversione a intero. E si
verifica che il controllo abbia POTENZA, cioe' che fallisca se la scala e' sbagliata.
"""
print("\n [SCALA] controllo positivo su daily_funding vs groupby pandas:")
for a in ASSETS:
ser = rate[a]["series"]
g = ser.groupby((ser.index - pd.Timedelta(seconds=1)).floor("D")).sum()
g = g[(g.index >= ser.index.min().floor("D") + pd.Timedelta("1D")) &
(g.index <= ser.index.max().floor("D") - pd.Timedelta("1D"))]
gi = pd.DatetimeIndex(g.index)
mine = daily_funding(rate[a], gi)
dmax = float(np.max(np.abs(mine - g.values)))
assert dmax < 1e-15, f"scala tempi rotta su {a}: max|dif|={dmax:.3e}"
bad = dict(ts=rate[a]["ts"] * 1000, cum=rate[a]["cum"]) # scala x1000 (us al posto di ms)
t = to_ms(gi[:200])
wrong = cumF(bad, t + 24 * MSH) - cumF(bad, t)
assert float(np.max(np.abs(wrong - mine[:200]))) > 1e-9, "il controllo NON ha potenza"
print(f" {a}: {len(g):,} giorni confrontati, max|dif| = {dmax:.3e} OK "
f"(e fallisce se la scala e' x1000 -> ha potenza)")
lo = min(rate[a]["series"].index.min() for a in ASSETS)
print(" [COPERTURA] il libro parte dal 2019-03-14, ma `get_funding_rate_history` serve una")
print(f" FINESTRA MOBILE di ~7,3 anni: oggi il piu' vecchio dato e' del {lo.date()}")
print(" (i perpetual INVERSE esistono dal 2018-08/2019-03: la storia c'era, il venue")
print(" non la serve piu'). -> i primi ~47 giorni del backtest sono senza funding")
print(" modellato = 1,7% del campione. Dichiarato, non stimato.")
# ===========================================================================
# 1) REPLICHE (funding spento => bit-exact) e 3) LE SERIE COL FUNDING
# ===========================================================================
def tp01_canonical(rate: dict | None) -> tuple[pd.Series, dict]:
"""TP01 canonico 50/50 (la lente di `al.tp01_baseline_daily`), con funding opzionale."""
tp = TrendPortfolio(**CANONICAL)
series, diag = {}, {}
for a in ASSETS:
df = al.get(a, "1d")
idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
net, _ = tp.net_returns(df)
net = np.asarray(net, float).copy()
tgt = np.asarray(tp.target_series(df), float)
pos = np.zeros(len(tgt)); pos[1:] = tgt[:-1] # tenuta durante la barra i
if rate is not None:
f = daily_funding(rate[a], idx)
net = np.clip(net - pos * f, -0.99, None)
net[0] = 0.0
diag[a] = dict(pos=pos, f=f, idx=idx)
series[a] = pd.Series(net, index=idx)
J = pd.concat(series, axis=1, join="inner").fillna(0.0)
return al._to_daily(0.5 * J["BTC"] + 0.5 * J["ETH"]), diag
def tp01_realistic(alloc: float, rate: dict | None) -> tuple[pd.Series, dict]:
"""TP01 con min-order $5 (la lente di `CC.tp01_realistic`), con funding opzionale."""
cap_asset = alloc / len(ASSETS)
tp = TrendPortfolio(**CANONICAL)
series, diag = {}, {}
for a in ASSETS:
df = resample_1d(load_data(a, "1h"))
idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
r = simple_returns(df["close"].values.astype(float))
tgt = np.nan_to_num(np.asarray(tp.target_series(df), float))
held = np.empty(len(tgt)); cur = 0.0
for i in range(len(tgt)):
if abs(tgt[i] - cur) * cap_asset >= CC.DERIBIT_MIN_ORDER:
cur = tgt[i]
held[i] = cur
pos = np.zeros(len(held)); pos[1:] = held[:-1]
net = pos * r - CC.DERIBIT_FEE_SIDE * np.abs(np.diff(pos, prepend=0.0))
fund = np.zeros(len(pos))
if rate is not None:
fund = pos * daily_funding(rate[a], idx)
net = net - fund
net[0] = 0.0
diag[a] = dict(pos=pos, fund=fund, idx=idx)
series[a] = pd.Series(np.clip(net, -0.99, None), index=idx)
J = pd.concat(series, axis=1, join="inner").fillna(0.0)
out = pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index)
if out.index.tz is None:
out.index = out.index.tz_localize("UTC")
return out, diag
def sim_equity_f(ltf: pd.DataFrame, ent: list, mode: str, ts5, c5,
rate: dict | None) -> tuple[np.ndarray, list]:
"""Replica di `r02.sim_equity` con UN grado di liberta' in piu': il funding accumulato
fra ingresso e uscita, col SEGNO della posizione (uno short lo INCASSA).
Con rate=None deve dare bit-exact `r02.sim_equity` — e' la replica di controllo."""
c = ltf["close"].values.astype(float)
h = ltf["high"].values.astype(float)
l = ltf["low"].values.astype(float)
n = len(c)
close_ts = ltf["timestamp"].values.astype(np.int64) + r02.MS_LTF
def px_hour(t: int) -> float:
hb = ((t + MSH - 1) // MSH) * MSH
j = np.searchsorted(ts5, hb, side="left")
return c5[min(j, len(c5) - 1)]
def t_hour(t: int) -> int:
return int(((t + MSH - 1) // MSH) * MSH)
initial = 1000.0
capital = initial
equity = np.full(n, capital, dtype=float)
busy_until = -1
trades = []
for i in range(n):
e = ent[i] if i < len(ent) else None
if e is None or e.get("dir", 0) == 0:
equity[i] = capital
continue
if i <= busy_until:
equity[i] = capital
continue
direction = int(e["dir"])
tp = e.get("tp"); sl = e.get("sl")
max_bars = int(e.get("max_bars") or 24)
entry = c[i] if mode in ("canonical", "barclose") else px_hour(close_ts[i])
exit_idx = min(i + max_bars, n - 1)
exit_lvl = c[exit_idx]
for j in range(i + 1, min(i + max_bars + 1, n)):
hit_sl = sl is not None and (
(direction == 1 and l[j] <= sl) or (direction == -1 and h[j] >= sl))
hit_tp = tp is not None and (
(direction == 1 and h[j] >= tp) or (direction == -1 and l[j] <= tp))
if hit_sl:
exit_lvl, exit_idx = sl, j
break
if hit_tp:
exit_lvl, exit_idx = tp, j
break
exit_lvl, exit_idx = c[j], j
if mode == "canonical":
exit_price = exit_lvl
elif mode == "barclose":
exit_price = c[exit_idx]
else:
exit_price = px_hour(close_ts[exit_idx])
gross = (exit_price - entry) / entry * direction
net = gross - 0.001
f_trade = 0.0
if rate is not None:
t_in = close_ts[i] if mode in ("canonical", "barclose") else t_hour(close_ts[i])
t_out = close_ts[exit_idx] if mode in ("canonical", "barclose") else t_hour(close_ts[exit_idx])
f_trade = float(cumF(rate, t_out) - cumF(rate, t_in)) * direction
net = net - f_trade
trades.append(dict(i=i, exit_idx=exit_idx, dir=direction, gross=gross,
f=f_trade, hours=(close_ts[exit_idx] - close_ts[i]) / MSH,
t_in=close_ts[i], t_out=close_ts[exit_idx]))
capital += capital * net
capital = max(capital, 1.0)
equity[i:exit_idx + 1] = capital
busy_until = exit_idx
last = initial
for k in range(n):
if equity[k] != last and equity[k] != initial:
last = equity[k]
else:
equity[k] = last
return equity, trades
_SKH: dict = {}
def skh_daily(a: str, off: int, mode: str, rate: dict | None) -> tuple[pd.Series, list]:
key = (a, off, mode, rate is not None)
if key in _SKH:
return _SKH[key]
_, m0, ltf, ent = r02.run_asset(a, off)
df5 = r02.get5m(a)
ts5 = df5["timestamp"].values.astype(np.int64) + r02.MS5
c5 = df5["close"].values.astype(float)
eq, tr = sim_equity_f(ltf, ent, mode, ts5, c5, rate[a] if rate else None)
s = pd.Series(eq, index=pd.DatetimeIndex(pd.to_datetime(ltf["datetime"], utc=True)))
d = s.resample("1D").last().ffill().pct_change().dropna()
_SKH[key] = (d, tr)
return _SKH[key]
def skh_book(off: int, mode: str, rate: dict | None) -> pd.Series:
J = pd.concat({a: skh_daily(a, off, mode, rate)[0] for a in ASSETS},
axis=1, join="inner").fillna(0.0)
return pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index)
def book_75_25(rate: dict | None, off: int = 0, lens: str = "hourly",
alloc: float = ALLOC) -> pd.Series:
"""Il libro LIVE: TP01 75 + SKH01 25, esattamente la lente di CC.book_series."""
skh = skh_book(off, lens, rate)
if skh.index.tz is None:
skh.index = skh.index.tz_localize("UTC")
tp, _ = tp01_realistic(alloc * 0.75, rate)
return combine_outer({"TP01": tp, "SKH01": skh}, {"TP01": 0.75, "SKH01": 0.25},
lo=pd.Timestamp("2019-03-14", tz="UTC"))
def sezione1(rate: dict) -> None:
sez("1) REPLICA DI CONTROLLO — con funding SPENTO ogni replica dev'essere BIT-EXACT")
ok = True
a0, _ = tp01_canonical(None)
b0 = al.tp01_baseline_daily()
d = float(np.max(np.abs(a0.values - b0.reindex(a0.index).values)))
ok &= d == 0.0
print(f" tp01_canonical(None) vs al.tp01_baseline_daily max|dif| = {d:.3e} "
f"{'OK' if d == 0 else 'FALLITA'}")
a1, _ = tp01_realistic(ALLOC * 0.75, None)
b1 = CC.tp01_realistic(ALLOC * 0.75)
d = float(np.max(np.abs(a1.values - b1.reindex(a1.index).values)))
ok &= d == 0.0
print(f" tp01_realistic(450,None) vs CC.tp01_realistic(450) max|dif| = {d:.3e} "
f"{'OK' if d == 0 else 'FALLITA'}")
for mode in ("canonical", "hourly"):
for a in ASSETS:
_, m0, ltf, ent = r02.run_asset(a, 0)
df5 = r02.get5m(a)
ts5 = df5["timestamp"].values.astype(np.int64) + r02.MS5
c5 = df5["close"].values.astype(float)
eqA, _ = sim_equity_f(ltf, ent, mode, ts5, c5, None)
eqB = r02.sim_equity(ltf, ent, mode, ts5_close=ts5, c5=c5)
d = float(np.max(np.abs(eqA - eqB)))
ok &= d == 0.0
print(f" sim_equity_f({a},{mode:<9}) vs r02.sim_equity "
f"max|dif| = {d:.3e} {'OK' if d == 0 else 'FALLITA'}")
a2 = book_75_25(None)
b2 = CC.book_series("hourly")
d = float(np.max(np.abs(a2.values - b2.reindex(a2.index).values)))
ok &= d == 0.0
print(f" book_75_25(None) vs CC.book_series('hourly') max|dif| = {d:.3e} "
f"{'OK' if d == 0 else 'FALLITA'}")
if not ok:
raise SystemExit("REPLICA FALLITA: nessun delta e' pubblicabile.")
print("\n ✅ 5/5 repliche bit-exact. Da qui in poi l'UNICO grado di liberta' e' il funding.")
# ===========================================================================
# 2) IL SEGNO — quanto si e' esposti, e da che lato
# ===========================================================================
def sezione2(rate: dict) -> dict:
sez("2) IL SEGNO — l'esposizione non e' assunta, e' appaiata col funding del suo strumento")
out = {}
_, dg = tp01_realistic(ALLOC * 0.75, rate)
print("\n TP01 (long-flat, posizione continua vol-targeted; esposizione = frazione del")
print(" capitale della GAMBA; nel libro pesa 0.75 x 0.5 per asset = 0.375 x posizione)")
print(f" {'asset':>6} {'giorni':>7} {'%tempo>0':>9} {'pos media':>10} {'pos med|>0':>11} "
f"{'funding pagato/anno':>20} {'esposiz. x tasso':>17}")
tot_leg = []
for a in ASSETS:
pos, f, idx = dg[a]["pos"], dg[a]["fund"], dg[a]["idx"]
yrs = len(pos) / 365.25
drag = float(f.sum()) / yrs # in frazione del capitale-gamba, per anno
rt = rate[a]["series"]
rt = rt[(rt.index >= idx.min()) & (rt.index <= idx.max())]
print(f" {a:>6} {len(pos):>7,} {float((pos > 0).mean()):>9.1%} {float(pos.mean()):>10.3f} "
f"{float(pos[pos > 0].mean()):>11.3f} {drag:>19.3%} "
f"{float(pos.mean()) * float(rt.mean()) * 24 * 365.25:>16.3%}")
tot_leg.append(drag)
out["tp01_leg_drag"] = float(np.mean(tot_leg))
print(f" -> drag medio per gamba {out['tp01_leg_drag']:.3%}/anno = sullo SLEEVE "
f"{out['tp01_leg_drag']:.3%}/anno (le due gambe pesano 0.5 ciascuna)")
print("\n SKH01 (long/short, round-trip discreti, non-overlap: sta a mercato una frazione")
print(" del tempo — e uno SHORT il funding lo INCASSA)")
print(f" {'asset':>6} {'trade':>6} {'%long':>7} {'%tempo a mercato':>17} {'ore/trade':>10} "
f"{'pagato/anno':>12} {'incassato/anno':>15} {'NETTO/anno':>11}")
tot_skh = []
for a in ASSETS:
d, tr = skh_daily(a, 0, "hourly", rate)
span_h = (tr[-1]["t_out"] - tr[0]["t_in"]) / MSH
yrs = span_h / (24 * 365.25)
f = np.array([t["f"] for t in tr])
dirs = np.array([t["dir"] for t in tr])
hrs = np.array([t["hours"] for t in tr])
inmkt = hrs.sum() / span_h
paid = float(f[f > 0].sum()) / yrs
got = float(-f[f < 0].sum()) / yrs
net = float(f.sum()) / yrs
tot_skh.append(net)
print(f" {a:>6} {len(tr):>6} {float((dirs > 0).mean()):>7.1%} {inmkt:>17.1%} "
f"{float(hrs.mean()):>10.1f} {paid:>11.3%} {got:>14.3%} {net:>10.3%}")
out["skh_drag"] = float(np.mean(tot_skh))
print(f" -> drag NETTO dello sleeve SKH01 {out['skh_drag']:.3%}/anno")
print("\n ❌ IPOTESI DEL BRIEF FALSIFICATA: «SKH01 sta a mercato solo il 12% del tempo ->")
print(" paga MOLTO MENO funding di TP01, ed e' un argomento a suo favore mai contato».")
print(f" Misurato: SKH01 {out['skh_drag']:.2%}/anno contro TP01 "
f"{out['tp01_leg_drag']:.2%}/anno di sleeve — SKH01 paga di PIU', non di meno.")
print(" Due ragioni, entrambe misurabili: (i) quando e' a mercato sta a nozionale 1.0x,")
print(" mentre TP01 vol-targeted sta a 0.14x in media -> a parita' di tempo pesa 7x;")
print(" (ii) entra sui breakout, e i breakout sono proprio quando il funding e' caro.")
print("\n 📌 IL MECCANISMO, ed e' il risultato trasferibile: ESPOSIZIONE e FUNDING sono")
print(" POSITIVAMENTE CORRELATI. Tasso medio annualizzato, non condizionato vs")
print(" condizionato all'essere in posizione:")
print(f" {'':>6} {'incondizionato':>15} {'| TP01 long':>12} {'| SKH01 a mercato':>18} "
f"{'rapporto TP01':>14} {'rapporto SKH':>13}")
for a in ASSETS:
ser = rate[a]["series"]
unc = float(ser.mean()) * 24 * 365.25
# TP01: media del tasso GIORNALIERO pesata dalla posizione (= tasso "che si paga davvero")
pos, f, idx = dg[a]["pos"], dg[a]["fund"], dg[a]["idx"]
fd = daily_funding(rate[a], idx)
w = pos / pos.sum() if pos.sum() > 0 else pos
tp_cond = float(np.sum(w * fd)) * 365.25
# SKH01: media del tasso orario sulle ore in cui e' a mercato (segno a parte)
_, tr = skh_daily(a, 0, "hourly", rate)
ore = float(np.sum([abs(t["hours"]) for t in tr]))
lordo = float(np.sum([abs(cumF(rate[a], t["t_out"]) - cumF(rate[a], t["t_in"]))
for t in tr]))
skh_cond = lordo / ore * 24 * 365.25 if ore else float("nan")
print(f" {a:>6} {unc:>15.2%} {tp_cond:>12.2%} {skh_cond:>18.2%} "
f"{tp_cond/unc:>13.2f}x {skh_cond/unc:>12.2f}x")
print(" -> il prodotto ingenuo «esposizione media x tasso medio» SOTTOSTIMA di ~2x. E' la")
print(" forma esatta della stima grezza da cui e' nato questo filone (14,8% x 0,103 =")
print(" 1,53%/anno): giusta come ordine di grandezza, sbagliata nei due fattori.")
print(" ⚠️ Il rapporto e' su TUTTO il campione e mescola due cose vere ma diverse: la")
print(" correlazione DENTRO l'anno (trend long = perp in contango) e il fatto che i")
print(" due anni di funding piu' caro (2019, 2021) sono anche quelli in cui TP01 era")
print(" piu' long. Per il COSTO non cambia nulla — si paga lo stesso — ma per")
print(" PREVEDERLO si', e la previsione non e' questo numero.")
print(f"\n 📌 A livello di LIBRO 75/25: 0.75 x {out['tp01_leg_drag']:.3%} + "
f"0.25 x {out['skh_drag']:.3%} = "
f"{0.75 * out['tp01_leg_drag'] + 0.25 * out['skh_drag']:.3%}/anno di costo lordo")
print(" (e' una scomposizione, non il risultato: il numero che conta e' il DELTA di")
print(" drift misurato sulla serie composta, sezione 3.)")
out["book_drag_decomp"] = 0.75 * out["tp01_leg_drag"] + 0.25 * out["skh_drag"]
return out
# ===========================================================================
# 3) I NUMERI
# ===========================================================================
def stat(s: pd.Series) -> dict:
m = metrics(s)
r = np.asarray(s.dropna().values, float)
return dict(full=al._sh(s), is_=al._sh(s[s.index < HOLDOUT]), hold=al._sh(s[s.index >= HOLDOUT]),
dd=al._dd_ret(s), cagr=m["cagr"], drift=float(r.mean()) * 365.25,
vol=float(r.std()) * np.sqrt(365.25), n=len(r))
def riga(nome: str, a: dict, b: dict) -> None:
print(f" {nome:<34} {a['full']:>7.3f} {a['is_']:>7.3f} {a['hold']:>7.3f} {a['dd']:>7.1%} "
f"{a['cagr']:>7.2%} {a['drift']:>7.2%}")
print(f" {' + funding':<34} {b['full']:>7.3f} {b['is_']:>7.3f} {b['hold']:>7.3f} "
f"{b['dd']:>7.1%} {b['cagr']:>7.2%} {b['drift']:>7.2%}")
print(f" {' DELTA':<34} {b['full']-a['full']:>+7.3f} {b['is_']-a['is_']:>+7.3f} "
f"{b['hold']-a['hold']:>+7.3f} {b['dd']-a['dd']:>+7.1%} {b['cagr']-a['cagr']:>+7.2%} "
f"{b['drift']-a['drift']:>+7.2%}\n")
def sezione3(rate: dict) -> dict:
sez("3) I NUMERI COL FUNDING DENTRO — netti fee 0.10% RT, hold-out 2025+")
print(f" {'serie':<34} {'ShFULL':>7} {'ShIS':>7} {'ShHOLD':>7} {'maxDD':>7} "
f"{'CAGR':>7} {'drift':>7}")
out = {}
p = {}
p["TP01 canonico (soffitto)"] = (tp01_canonical(None)[0], tp01_canonical(rate)[0])
p[f"TP01 realistic ${ALLOC*0.75:.0f}"] = (tp01_realistic(ALLOC * 0.75, None)[0],
tp01_realistic(ALLOC * 0.75, rate)[0])
p["SKH01 50/50 canonical"] = (skh_book(0, "canonical", None), skh_book(0, "canonical", rate))
p["SKH01 50/50 path live"] = (skh_book(0, "hourly", None), skh_book(0, "hourly", rate))
p["LIBRO 75/25 (path live)"] = (book_75_25(None), book_75_25(rate))
for k, (A, B) in p.items():
sa, sb = stat(A), stat(B)
riga(k, sa, sb)
out[k] = (sa, sb)
print(" Per ANNO, libro 75/25 (drift lordo -> netto funding). La colonna ESPOSIZIONE serve")
print(" a non confondere 'il funding e' sceso' con 'il libro era flat': sono cose diverse e")
print(" producono lo stesso delta piccolo.")
A, B = p["LIBRO 75/25 (path live)"]
_, dgx = tp01_realistic(ALLOC * 0.75, rate)
expo = pd.concat({a: pd.Series(dgx[a]["pos"], index=dgx[a]["idx"]) for a in ASSETS},
axis=1, join="inner").mean(axis=1)
print(f" {'anno':>6} {'senza':>9} {'con':>9} {'delta':>9} {'funding annualiz.':>18} "
f"{'esposiz. TP01':>14}")
for y in sorted(set(A.index.year)):
ga, gb = A[A.index.year == y], B[B.index.year == y]
if len(ga) < 30:
continue
da, db = float(ga.mean()) * 365.25, float(gb.mean()) * 365.25
fr = np.mean([rate[a]["series"][rate[a]["series"].index.year == y].mean()
for a in ASSETS]) * 24 * 365.25
ex = float(expo[expo.index.year == y].mean())
print(f" {y:>6} {da:>9.2%} {db:>9.2%} {db-da:>+9.2%} {fr:>17.2%} {ex:>14.3f}")
rec = pd.Timestamp("2025-01-01", tz="UTC")
dA, dB = float(A[A.index >= rec].mean()) * 365.25, float(B[B.index >= rec].mean()) * 365.25
print(f" {'2025+':>6} {dA:>9.2%} {dB:>9.2%} {dB-dA:>+9.2%} "
f"{np.mean([rate[a]['series'][rate[a]['series'].index >= rec].mean() for a in ASSETS])*24*365.25:>17.2%} "
f"{float(expo[expo.index >= rec].mean()):>14.3f}")
print(" 📌 IL COSTO NON E' STAZIONARIO, e le due cause si LEGGONO nelle due colonne: il")
print(" funding Deribit si e' compresso (16,9% nel 2021 -> 1,9% nel 2026) E il libro sta")
print(" meno esposto. Chi cita UN numero per il funding cita una media su un regime")
print(" cambiato; chi cita il numero del 2026 sta citando un libro quasi flat.")
# finestra SOLO-LINEARE (nessun proxy): il controllo che il numero non venga dal proxy
lo = rate["BTC"]["switch"]
print(f"\n CONTROLLO — stessa misura sulla sola finestra LINEARE (>= {lo.date()},")
print(" nessun proxy inverse):")
print(f" {'serie':<34} {'d ShFULL':>9} {'d CAGR':>9} {'d drift':>9}")
for k, (A, B) in p.items():
A2, B2 = A[A.index >= lo], B[B.index >= lo]
sa, sb = stat(A2), stat(B2)
print(f" {k:<34} {sb['full']-sa['full']:>+9.3f} {sb['cagr']-sa['cagr']:>+9.2%} "
f"{sb['drift']-sa['drift']:>+9.2%}")
out["_pairs"] = p
return out
# ===========================================================================
# 4) LE DECISIONI
# ===========================================================================
def sezione4a(res3: dict) -> None:
sez("4a) IL SOFFITTO ~1.3 SI ABBASSA?")
a, b = res3["TP01 canonico (soffitto)"]
print(f" TP01 canonico (l'unica serie a cui il 'soffitto ~1.3' si riferisce):")
print(f" Sharpe FULL {a['full']:.3f} -> {b['full']:.3f} ({b['full']-a['full']:+.3f})")
print(f" CAGR {a['cagr']:.2%} -> {b['cagr']:.2%}")
print(f" Il soffitto e' un numero arrotondato a UNA cifra: si abbassa di "
f"{abs(b['full']-a['full']):.3f}, quindi resta '~1.3'"
if abs(b["full"] - a["full"]) < 0.05 else
f" Il soffitto si sposta di {b['full']-a['full']:+.3f}: va RIDETTO.")
print(" ⚠️ Ma la lettura utile e' l'altra: il soffitto e' una soglia di SELEZIONE, e ogni")
print(" candidato direzionale mai valutato in questo progetto e' stato misurato SENZA")
print(" funding. Il confronto candidato-vs-TP01 e' quindi appaiato (entrambi al lordo)")
print(" per i candidati LONG/SHORT a esposizione simile, e distorto a favore di chi sta")
print(" a mercato di piu'. Chi paga di piu' e' chi e' piu' esposto piu' a lungo.")
def sezione4b(rate: dict, offsets: tuple) -> None:
sez("4b) IL PESO 75/25 SI MUOVE? (il funding e' un argomento a favore di SKH01 mai contato)")
print(" L'ipotesi era: «SKH01 sta a mercato una frazione del tempo e puo' incassare da")
print(" short -> paga MENO funding di TP01, quindi il peso ottimale sale». La prima meta'")
print(" e' gia' FALSIFICATA in sez. 2 (SKH01 paga di piu' per dollaro di sleeve). Resta la")
print(" seconda, che non e' implicata dalla prima: il peso si sposta?")
print(f" Sweep w in {list(skl.W_GRID)} x {len(offsets)} offset della griglia 230m/690m.")
print(" Statistica = MEDIANA DELLE DIFFERENZE APPAIATE (regola del progetto), lente")
print(" 'hourly' = path live, selezione sullo Sharpe IN-SAMPLE come il 24/07.\n")
B0, _ = tp01_canonical(None)
B1, _ = tp01_canonical(rate)
def blend(B, skh, w):
if w == 0.0:
return B
bl = combine_outer({"TP": B, "SKH": skh}, {"TP": 1.0 - w, "SKH": w})
return bl[bl.index >= B.index.min()]
S = {False: {}, True: {}}
for off in offsets:
S[False][off] = skh_book(off, "hourly", None)
S[True][off] = skh_book(off, "hourly", rate)
print(f" {'w':>6} | {'IS mediano':>11} {'IS med +f':>11} {'d IS':>8} {'>0':>5} | "
f"{'FULL med':>9} {'FULL +f':>9} {'d FULL':>8} | {'HOLD med':>9} {'HOLD +f':>9}")
rows = {}
for w in skl.W_GRID:
vI0, vI1, vF0, vF1, vH0, vH1 = [], [], [], [], [], []
for off in offsets:
b0 = blend(B0, S[False][off], w)
b1 = blend(B1, S[True][off], w)
vI0.append(al._sh(b0[b0.index < HOLDOUT])); vI1.append(al._sh(b1[b1.index < HOLDOUT]))
vF0.append(al._sh(b0)); vF1.append(al._sh(b1))
vH0.append(al._sh(b0[b0.index >= HOLDOUT])); vH1.append(al._sh(b1[b1.index >= HOLDOUT]))
dI = np.array(vI1) - np.array(vI0)
dF = np.array(vF1) - np.array(vF0)
rows[w] = dict(is0=np.median(vI0), is1=np.median(vI1), dI=np.median(dI),
f0=np.median(vF0), f1=np.median(vF1), dF=np.median(dF),
h0=np.median(vH0), h1=np.median(vH1), pos=float((dI > 0).mean()))
r_ = rows[w]
print(f" {w:>6.2f} | {r_['is0']:>11.3f} {r_['is1']:>11.3f} {r_['dI']:>+8.3f} "
f"{r_['pos']:>5.0%} | {r_['f0']:>9.3f} {r_['f1']:>9.3f} {r_['dF']:>+8.3f} | "
f"{r_['h0']:>9.3f} {r_['h1']:>9.3f}")
w0 = max(skl.W_GRID, key=lambda w: rows[w]["is0"])
w1 = max(skl.W_GRID, key=lambda w: rows[w]["is1"])
print(f"\n argmax dello Sharpe IS mediano: SENZA funding w* = {w0:.2f} "
f"CON funding w* = {w1:.2f}")
print(f" peso live = 0.25. Regola a priori del 24/07: a parita' (<{skl.TIE_TOL}) si sta al")
print(f" piu' vicino a 0.25. |IS(w*) - IS(0.25)| con funding = "
f"{abs(rows[w1]['is1'] - rows[0.25]['is1']):.3f}")
if w0 == w1:
print(" 📌 Il funding NON sposta l'argmax: il vantaggio di SKH01 e' reale ma piccolo")
print(" rispetto alla larghezza del plateau. Peso INVARIATO.")
else:
print(f" 📌 L'argmax si sposta {w0:.2f} -> {w1:.2f}: da verificare con weights_tilt_null.")
def sezione4c(rate: dict) -> dict:
sez("4c) I MURI DI CAPITALE E LA RENDITA PERPETUA (vivono sul DRIFT, non sullo Sharpe)")
A, B = book_75_25(None), book_75_25(rate)
need_gross = CC.TARGET_EUR_DAY * 365.0 * EURUSD / (1 - CC.TAX_RATE)
print(f" bersaglio: 50 EUR/g netti = ${need_gross:,.0f}/anno lordi (aliquota "
f"{CC.TAX_RATE:.0%}), stessa convenzione di r0726_capwall_refresh.")
print(f" de-luck del drift x{DELUCK} (MISURATO, r0726_deluck_factor). MC {N_PATHS_WALL} path")
print(" -> risoluzione ~1% sul muro: si citano 3 cifre significative, non il dollaro.\n")
print(f" {'serie':<28} {'drift':>8} {'vol':>7} {'perpetua':>9} {'muro':>12} {'leva':>5}")
out = {}
for nome, s in (("senza funding", A), ("con funding", B)):
r = s.values.astype(float)
r_dl = r - (1.0 - DELUCK) * float(r.mean())
for k in (1.0, 1.5):
_, perp, cap = WR.perp_and_wall(r_dl, k, need_gross)
out[(nome, k)] = (perp, cap)
print(f" {nome + f' x{DELUCK}':<28} {float(r_dl.mean())*365.25:>8.2%} "
f"{float(r_dl.std())*np.sqrt(365.25):>7.2%} {perp:>9.2%} ${cap:>11,.0f} {k:>5.2f}")
p0, c0 = out[("senza funding", 1.0)]
p1, c1 = out[("con funding", 1.0)]
print(f"\n 📌 A leva 1.0: perpetua {p0:.2%} -> {p1:.2%}; muro ${c0:,.0f} -> ${c1:,.0f} "
f"({(c1/c0-1):+.1%})")
print(f" Il muro pubblicato e' $272k (lordo) / $258k (netto fisco). Qui si misura il")
print(f" DELTA con la stessa macchineria, non un muro nuovo.")
print("\n TRAIETTORIA da $600 al capitale-rendita (mediana anni / P entro 20a), stesso")
print(" bootstrap a blocchi di r0726_capwall_refresh.trajectory:")
print(f" {'EUR/mese':>9} | {'senza funding':>22} | {'con funding':>22}")
r0 = A.values.astype(float); r0 = r0 - (1 - DELUCK) * r0.mean()
r1 = B.values.astype(float); r1 = r1 - (1 - DELUCK) * r1.mean()
t0 = WR.trajectory(r0, c0, n_paths=2000)
t1 = WR.trajectory(r1, c1, n_paths=2000)
for dep in (0, 250, 500, 1_000, 2_000):
a_, b_ = t0[dep], t1[dep]
fa = f"{a_['p50']:>6.1f}a P20 {a_['p20y']:>5.0%}" if a_["p50"] == a_["p50"] else f"{'mai':>16}"
fb = f"{b_['p50']:>6.1f}a P20 {b_['p20y']:>5.0%}" if b_["p50"] == b_["p50"] else f"{'mai':>16}"
print(f" {dep:>9} | {fa:>22} | {fb:>22}")
return out
def g_exact(r: np.ndarray, k: float) -> float:
x = 1.0 + k * r
return float(np.mean(np.log(np.maximum(x, 1e-12)))) * 365.25
def k_star(r: np.ndarray, kmax: float = 40.0, step: float = 0.05) -> tuple[float, float]:
ks = np.arange(step, kmax + step, step)
g = np.array([g_exact(r, k) for k in ks])
j = int(np.argmax(g))
return float(ks[j]), float(g[j])
def sezione4d(rate: dict) -> None:
sez("4d) k* E IL GRADINO DI LEVA — il funding SCALA con la leva")
print(" Il funding e' proporzionale al NOZIONALE: a leva k si paga k volte tanto. Quindi")
print(" la serie col funding va scalata per k come tutto il resto (il costo non e' fisso),")
print(" e il gradino 1.00x -> 1.25/1.50x misurato ieri ne mangia una parte.\n")
A, B = book_75_25(None), book_75_25(rate)
print(f" {'serie':<22} {'drift':>8} {'k*':>7} {'g(k*)':>8} {'g(1.00)':>9} {'g(1.25)':>9} "
f"{'g(1.50)':>9} {'mezzo-Kelly':>12}")
for nome, s in (("senza funding", A), ("con funding", B)):
r = s.values.astype(float)
ks, gs = k_star(r)
print(f" {nome:<22} {float(r.mean())*365.25:>8.2%} {ks:>7.2f} {gs:>8.2%} "
f"{g_exact(r, 1.0):>9.2%} {g_exact(r, 1.25):>9.2%} {g_exact(r, 1.50):>9.2%} "
f"{ks/2:>12.2f}")
r_dl = r - (1 - DELUCK) * r.mean()
ks2, gs2 = k_star(r_dl)
print(f" {' (de-luck x0.89)':<22} {float(r_dl.mean())*365.25:>8.2%} {ks2:>7.2f} "
f"{gs2:>8.2%} {g_exact(r_dl, 1.0):>9.2%} {g_exact(r_dl, 1.25):>9.2%} "
f"{g_exact(r_dl, 1.50):>9.2%} {ks2/2:>12.2f}")
rA = A.values.astype(float); rA -= (1 - DELUCK) * rA.mean()
rB = B.values.astype(float); rB -= (1 - DELUCK) * rB.mean()
gain0 = g_exact(rA, 1.25) - g_exact(rA, 1.0)
gain1 = g_exact(rB, 1.25) - g_exact(rB, 1.0)
print(f"\n 📌 Guadagno in log-crescita del gradino 1.00 -> 1.25 (de-luckato):")
print(f" senza funding {gain0:+.2%}/anno · con funding {gain1:+.2%}/anno "
f"(-{(1-gain1/gain0)*100:.0f}% del guadagno)" if gain0 > 0 else "")
print(" Il gradino resta positivo finche' k* > 1.25 con il funding dentro.")
# ===========================================================================
# 5) CONTROLLO SUL CONTO VERO
# ===========================================================================
RE_HDR = re.compile(r"^===== (\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z) cron_book")
RE_POS = re.compile(r"^\s+(BTC|ETH) TP .*?\|\s*pos \$([+-][\d,\.]+)")
def leggi_posizioni_live() -> pd.DataFrame:
"""Posizione oraria REALE per asset, ricostruita da logs/cron_book.log."""
rows, ts = [], None
for ln in (ROOT / "logs" / "cron_book.log").read_text(errors="replace").splitlines():
m = RE_HDR.match(ln)
if m:
ts = pd.Timestamp(m.group(1))
continue
m = RE_POS.match(ln)
if m and ts is not None:
rows.append(dict(ts=ts, asset=m.group(1),
pos=float(m.group(2).replace(",", "").replace("+", ""))))
return pd.DataFrame(rows)
RE_ORD = re.compile(r"^\s+-> (BUY|SELL) ")
RE_LEG = re.compile(r"^\s+(BTC|ETH) TP ")
def ordini_sessione(settl: pd.Timestamp) -> dict:
"""Ordini REALMENTE inviati per asset dalla data `settl` (righe '-> BUY/SELL' del log,
che seguono la riga della gamba a cui appartengono)."""
out: dict = {}
ts, leg = None, None
for ln in (ROOT / "logs" / "cron_book.log").read_text(errors="replace").splitlines():
m = RE_HDR.match(ln)
if m:
ts = pd.Timestamp(m.group(1)); leg = None
continue
m = RE_LEG.match(ln)
if m:
leg = m.group(1)
continue
m = RE_ORD.match(ln)
if m and leg and ts is not None and ts >= settl:
b, s_ = out.get(leg, (0, 0))
out[leg] = (b + (m.group(1) == "BUY"), s_ + (m.group(1) == "SELL"))
return out
def sezione5(rate: dict) -> None:
sez("5) CONTROLLO SUL CONTO VERO — l'unico che non passa da un'assunzione")
df = leggi_posizioni_live()
if not len(df):
print(" logs/cron_book.log non leggibile o formato cambiato -> NON MISURATO.")
return
print(f" {len(df):,} osservazioni orarie di posizione ricostruite dal log del cron,")
print(f" {df['ts'].min()} -> {df['ts'].max()}")
print(" ⚠️ `data/live/book_executions.jsonl` ha `ts_utc` a 00:00:00 su 18/18 righe (difetto")
print(" gia' noto, ondata 22/08 §5): l'ora vera della posizione sta SOLO nel log del cron.")
def paid(g: pd.DataFrame, a: str) -> float:
t = to_ms(pd.DatetimeIndex(g["ts"]))
return float((g["pos"].values * (cumF(rate[a], t + MSH) - cumF(rate[a], t))).sum())
tot = {}
for a in ASSETS:
g = df[df["asset"] == a].sort_values("ts")
tot[a] = paid(g, a)
print(f" {a}: {len(g):>4} ore, posizione media ${g['pos'].mean():>7.2f} "
f"(a mercato {float((g['pos'].values != 0).mean()):.0%}, max ${g['pos'].max():.0f})"
f" -> funding modellato ${tot[a]:+.4f}")
print(f" TOTALE funding modellato da quando il libro esegue: ${sum(tot.values()):+.4f}")
print(" (a $635 di conto e posizioni da $40-190: e' un numero da CENTESIMI. La potenza del")
print(" controllo non viene dalla taglia, viene dal fatto che sia CONFRONTABILE.)")
print("\n Cio' che il conto RIPORTA (sola lettura via cerbero-mcp, nessun ordine):")
try:
from src.live.deribit import DeribitRead
ps = DeribitRead().positions("USDC")
except Exception as e:
print(f" conto non leggibile ({type(e).__name__}) -> confronto NON MISURATO.")
return
rp = {}
for p_ in ps:
inst = str(p_.get("instrument", ""))
a = "BTC" if inst.startswith("BTC") else "ETH"
rp[a] = float(p_.get("realized_pnl", 0.0))
print(f" {inst:<22} size ${float(p_.get('size', 0)):>8.2f} "
f"realized_pnl {rp[a]:>+9.6f} unrealized {float(p_.get('unrealized_pnl', 0)):>+9.6f}")
print("\n QUALE finestra? Deribit REGOLA i perpetual una volta al giorno alle 08:00 UTC.")
print(" Se `realized_pnl` e' il realizzato di SESSIONE, dev'essere il funding dall'ultimo")
print(" settlement — e allora la finestra giusta si riconosce perche' le altre NON tornano:")
now = pd.Timestamp.now(tz="UTC")
settl = now.normalize() + pd.Timedelta(hours=8)
if settl > now:
settl -= pd.Timedelta("1D")
fin = {a: df[df["asset"] == a].sort_values("ts") for a in ASSETS}
apert = {}
for a in ASSETS:
g = fin[a].reset_index(drop=True)
z = np.where(g["pos"].values == 0)[0]
apert[a] = g["ts"].iloc[int(z[-1])] if len(z) else g["ts"].iloc[0]
print(f" {'finestra':<34} {'BTC modello':>13} {'BTC conto':>11} {'err':>8} | "
f"{'ETH modello':>13} {'ETH conto':>11} {'err':>8}")
for lbl, lo in ((f"dal settlement {settl:%m-%d %H:%M} UTC", {a: settl for a in ASSETS}),
("dall'apertura della posizione", apert),
("ultime 24h", {a: now - pd.Timedelta("24h") for a in ASSETS}),
("da quando il libro esegue", {a: fin[a]["ts"].min() for a in ASSETS})):
cells = []
for a in ASSETS:
g = fin[a][pd.DatetimeIndex(fin[a]["ts"]) >= lo[a]]
m = paid(g, a)
err = (m / abs(rp[a]) - 1) if rp.get(a) else float("nan")
cells.append(f"{m:>13.6f} {rp.get(a, float('nan')):>11.6f} {err:>+7.1%}")
print(f" {lbl:<34} {cells[0]} | {cells[1]}")
# (1) va MISURATO sugli ORDINI VERI, non sulle variazioni del nozionale.
# ⚠️ ERRORE CATTURATO IN SESSIONE: la prima stesura contava `diff(pos) < 0` come
# "vendite" e ne trovava 3 per asset. NON erano vendite: `pos` e' il valore in USD di
# una posizione a taglia costante, e scende da solo quando il mark scende. Su un
# perpetual lineare il nozionale si muove SEMPRE, anche a mano ferma.
ordini = ordini_sessione(settl)
print("\n 📌 IDENTIFICAZIONE, e non e' un accordo di grandezza — e' un'ELIMINAZIONE:")
print(f" (1) ordini realmente inviati nella sessione (righe '-> BUY/SELL' del log, "
f"dalle {settl:%H:%M} UTC): "
+ " · ".join(f"{a} {ordini.get(a, (0, 0))[0]} buy / {ordini.get(a, (0, 0))[1]} sell"
for a in ASSETS))
if sum(v[1] for v in ordini.values()) == 0:
print(" ZERO vendite -> nessuna chiusura, nemmeno parziale -> il P&L di TRADING")
print(" realizzato di sessione e' ESATTAMENTE ZERO, e qualunque valore non nullo")
print(" di `realized_pnl` non puo' essere P&L di trading.")
else:
print(" ⚠️ Ci sono state VENDITE: parte di `realized_pnl` puo' essere P&L di")
print(" trading e l'eliminazione (1) NON vale in questa sessione. Restano (2)(3).")
print(" (2) il segno e' NEGATIVO su entrambe le gambe, che sono entrambe LONG: un long")
print(" in contango PAGA. E' il segno che il funding deve avere, non un caso.")
print(" (3) la finestra 'dal settlement' torna; 'dall'apertura' sbaglia di ~2-3x e")
print(" '24h' di ~15%. Il controllo ha POTENZA: le finestre sbagliate si vedono.")
print(" -> `realized_pnl` del proxy = realizzato di SESSIONE Deribit, che su una")
print(" posizione non chiusa E' il funding. Il modello di questo script e' quindi")
print(" VALIDATO contro il conto reale, non solo contro se stesso.")
print(" ⚠️ IL RESIDUO, misurato invece che raccontato. Il modello SOTTOSTIMA su entrambe le")
print(" gambe. Espresso in ORE-equivalenti all'ultimo tasso orario osservato:")
for a in ASSETS:
g = fin[a][pd.DatetimeIndex(fin[a]["ts"]) >= settl]
res = abs(rp.get(a, 0.0)) - paid(g, a)
ult = float(rate[a]["series"].iloc[-1]) * float(g["pos"].iloc[-1])
print(f" {a}: residuo ${res:+.6f} = {res/ult if ult else float('nan'):+.2f} ore "
f"di funding all'ultimo tasso (${ult:.6f}/ora)")
print(" Meccanismo candidato e coerente col segno su 2/2: il conto accumula in CONTINUO,")
print(" l'ultimo record orario e' delle 22:00 e la lettura e' delle 22:1x -> i minuti")
print(" dopo l'ultimo record non sono modellati. NON e' verificato che spieghi TUTTO il")
print(" residuo di BTC: si riporta la taglia, non una storia.")
print(" ⚠️ Limite: e' UNA sessione, su due posizioni. Non e' una serie storica di")
print(" pagamenti di funding — quella il conto non la espone via questo proxy.")
# ===========================================================================
# 6) COSA NON E' CATTURATO
# ===========================================================================
def sezione6(rate: dict) -> None:
sez("6) COSA NON E' CATTURATO — dichiarato e, dove si puo', quantificato")
print(" (a) L'approssimazione 'esposizione MEDIA x tasso MEDIO' contro l'accrual ESATTO.")
print(" Su TP01 non c'e' differenza (posizione continua, accrual giornaliero esatto).")
print(" Su SKH01, che entra ed esce, la differenza si MISURA:")
print(f" {'asset':>6} {'accrual esatto/anno':>20} {'media x media/anno':>20} "
f"{'errore':>9} {'errore rel.':>12}")
for a in ASSETS:
_, tr = skh_daily(a, 0, "hourly", rate)
span_h = (tr[-1]["t_out"] - tr[0]["t_in"]) / MSH
yrs = span_h / (24 * 365.25)
esatto = float(np.sum([t["f"] for t in tr])) / yrs
# approssimazione: (ore a mercato x direzione media) x tasso medio del periodo
hrs = np.array([t["hours"] for t in tr]); dirs = np.array([t["dir"] for t in tr])
rt = rate[a]["series"]
rt = rt[(rt.index >= pd.Timestamp(tr[0]["t_in"], unit="ms", tz="UTC")) &
(rt.index <= pd.Timestamp(tr[-1]["t_out"], unit="ms", tz="UTC"))]
approx = float((hrs * dirs).sum() * rt.mean()) / yrs
err = esatto - approx
print(f" {a:>6} {esatto:>19.4%} {approx:>19.4%} {err:>+8.4%} "
f"{(err/abs(esatto) if esatto else float('nan')):>11.1%}")
print(" -> l'approssimazione media-x-media NON e' innocua su uno sleeve che sceglie")
print(" QUANDO stare a mercato: e' la stessa forma dell'errore che il progetto ha")
print(" gia' pagato altrove (una media appaiata non e' la media delle medie).")
print(" Qui il numero pubblicato usa l'ACCRUAL ESATTO per trade.")
sw = min(rate[a]["switch"] for a in ASSETS)
b0 = pd.Timestamp("2019-03-14", tz="UTC")
tot = (rate["BTC"]["series"].index.max() - b0).days
print(f"\n (b) Il PROXY inverse prima del {sw.date()} ({(sw-b0).days/tot:.0%} del campione).")
print(" L'accordo misurato")
print(" in sezione 0 e' la banda dell'errore; la sezione 3 riporta anche la finestra")
print(" SOLO-LINEARE, dove il proxy non entra.")
print("\n (c) LA DISCRETEZZA. Deribit accumula il funding in CONTINUO (interest_1h e' gia'")
print(" il realizzato dell'ora): su questo venue non c'e' il problema 'posizione")
print(" all'istante del pagamento' che hanno i venue a 8h. Cio' che resta discreto e'")
print(" l'istante di USCITA di SKH01 dentro la barra 230m: il modello lo pone alla")
print(" CHIUSURA della barra di trigger (limite SUPERIORE del funding pagato).")
print(" Banda misurata sostituendo l'apertura della barra di trigger:")
for a in ASSETS:
_, tr = skh_daily(a, 0, "hourly", rate)
span_h = (tr[-1]["t_out"] - tr[0]["t_in"]) / MSH
yrs = span_h / (24 * 365.25)
hi = float(np.sum([t["f"] for t in tr])) / yrs
lo_ = float(np.sum([
(cumF(rate[a], t["t_out"] - r02.MS_LTF) - cumF(rate[a], t["t_in"])) * t["dir"]
for t in tr])) / yrs
print(f" {a}: [{lo_:.4%}, {hi:.4%}] /anno di sleeve (ampiezza {hi-lo_:.4%})")
print("\n (d) NON e' modellato il funding sul collaterale (il conto e' USDC, il margine non")
print(" rende) ne' il costo di prestito: non esistono su questo percorso.")
print(" (e) Il libro di RICERCA a 5 sleeve non e' ricalcolato qui: VRP01 e' opzioni (niente")
print(" funding), GTAA01 e' azioni (niente funding, e non e' deployabile), XS01 e XSR01")
print(" sono su Hyperliquid e il loro funding e' un'ALTRA misura, market-neutral per")
print(" costruzione (long e short si compensano solo se il funding e' uniforme).")
# ===========================================================================
def main() -> None:
print("=" * 100)
print("r0822d — IL FUNDING DEI PERPETUAL: il costo che il progetto non ha mai prezzato")
print(f"branch research/wave-0822 · {pd.Timestamp.now(tz='UTC')} · rete={'SI' if USE_NET else 'NO'}")
print("=" * 100)
d0 = sezione0()
rate = build_rate(d0["F"])
print(f"\n serie oraria unificata per asset (linear dove esiste, inverse prima):")
for a in ASSETS:
print(f" {a}: {len(rate[a]['ts']):,} ore, switch inverse->linear il "
f"{rate[a]['switch'].date()}")
verifica_scala(rate, d0["F"])
print(f"\n{clock()} sezione 1...")
sezione1(rate)
print(f"\n{clock()} sezione 2...")
d2 = sezione2(rate)
print(f"\n{clock()} sezione 3...")
d3 = sezione3(rate)
print(f"\n{clock()} sezione 4...")
sezione4a(d3)
sezione4b(rate, r02.OFFSETS)
sezione4c(rate)
sezione4d(rate)
print(f"\n{clock()} sezione 5...")
sezione5(rate)
sezione6(rate)
sez("GATE — esito di ciascuno, e quelli non girati DICHIARATI")
A0, B0 = book_75_25(None), book_75_25(rate)
ns = int(sum(len(skh_daily(a, 0, "hourly", rate)[1]) for a in ASSETS))
npos = int(sum(np.sum(np.array([t["gross"] for t in skh_daily(a, 0, "hourly", rate)[1]]) <= 0)
for a in ASSETS))
ip = al.implausible_sharpe(B0, n_trades=ns, n_losing_trades=npos)
print(f" implausible_sharpe : implausible={ip['implausible']} "
f"(Sharpe {ip['sharpe']:.2f}, maxDD {ip['maxdd']:.1%}, Calmar {ip['calmar']:.2f}, "
f"barre attive in perdita {ip['loss_frac']:.1%}) -> {ip['reasons'] or 'nessuna firma'}")
print(" replica bit-exact : PASSATA 5/5 (sez. 1) — nessun delta stampato prima.")
print(" controllo di scala tempi: PASSATO, con potenza verificata (sez. 0).")
print(" conto reale : PASSATO — modello vs `realized_pnl` di sessione, sez. 5.")
print(" anchor (banda d'ancora) : GIRATO sul peso — 23 offset, mediana delle differenze")
print(" APPAIATE (sez. 4b). Le altre tabelle sono all'ancora")
print(" canonica: il funding e' un COSTO deterministico, non una")
print(" selezione, e il suo delta e' negativo a 23/23 offset.")
print(" marginal_vs_tp01 / DSR / study_family_honest / de-levering null / causality:")
print(" NON GIRATI, e non per budget: qui non c'e' nessun")
print(" CANDIDATO. Non si propone una strategia, si toglie un")
print(" costo che mancava a quelle esistenti. Non c'e' una")
print(" griglia su cui barare e non c'e' un segnale di cui")
print(" verificare la causalita' (il funding di un'ora si")
print(" applica alla posizione tenuta in quell'ora).")
print(" weights_tilt_null : NON GIRATO perche' non serve — l'argmax del peso NON si")
print(" sposta (sez. 4b), quindi non c'e' nessun cambio di pesi")
print(" da sottoporre al gate.")
sez("SINTESI")
a, b = d3["LIBRO 75/25 (path live)"]
print(f" LIBRO 75/25 path live: drift {a['drift']:.2%} -> {b['drift']:.2%} "
f"({b['drift']-a['drift']:+.2%}/anno)")
print(f" Sharpe FULL {a['full']:.3f} -> {b['full']:.3f}, "
f"HOLD {a['hold']:.3f} -> {b['hold']:.3f}, maxDD {a['dd']:.1%} -> {b['dd']:.1%}")
print(f" quota del drift bruciata dal funding: {(a['drift']-b['drift'])/a['drift']:.1%}")
print(f" decomposizione (sez.2): {d2['book_drag_decomp']:.3%}/anno")
rec = pd.Timestamp("2025-01-01", tz="UTC")
A_, B_ = d3["_pairs"]["LIBRO 75/25 (path live)"]
dr = (float(A_[A_.index >= rec].mean()) - float(B_[B_.index >= rec].mean())) * 365.25
lo_ = min(rate[x]["switch"] for x in ASSETS)
A2, B2 = A_[A_.index >= lo_], B_[B_.index >= lo_]
dl = (float(A2.mean()) - float(B2.mean())) * 365.25
print(f" ... {dl:.2%}/anno sulla sola finestra dello strumento VERO (>= {lo_.date()}, senza")
print(f" proxy), e {dr:.2%}/anno sul solo 2025+. IL NUMERO VA CITATO CON LA FINESTRA:")
print(f" la media di 7 anni non e' l'aspettativa di domani.")
print("\n COSA CAMBIA e COSA NO:")
print(" CAMBIA · il muro di capitale e la rendita perpetua (+17-18% di muro: la rendita")
print(" vive sul DRIFT, e il funding e' una tassa sul drift)")
print(" · la traiettoria da $600 (a EUR250/mese, P(entro 20a) 90% -> 64%)")
print(" · il soffitto direzionale BTC/ETH: ~1,31 -> ~1,15 netto funding")
print(" NON CAMBIA · il peso 75/25 (argmax invariato a 23/23 offset)")
print(" · il gradino di leva 1,00 -> 1,25x (k* 13,7 -> 12,3, guadagno -12%)")
print(" · il maxDD (+0,3pp: e' una tassa sul drift, non sulla coda)")
print(" · l'ordine delle leve del piano: versare resta la prima di tutte")
print("\n GRIGLIA DICHIARATA (contata al RIALZO):")
print(" 4 strumenti scaricati x ~55-90 richieste = ~290 chiamate al venue")
print(" 3 ipotesi sulla convenzione temporale · 2 lenti (merged / solo-lineare)")
print(" 5 serie x 2 (con/senza funding) = 10 valutazioni in sez. 3, + 8 anni di split")
print(" 8 pesi x 23 offset x 2 = 368 valutazioni di blend in sez. 4b")
print(" 4 muri x 26 passi di bisezione x 2.000 path MC + 2 x 5 livelli di traiettoria")
print(" 2 (con/senza) x 800 valori di k in sez. 4d · 4 finestre x 2 gambe in sez. 5")
print(" NESSUNA di queste e' una SELEZIONE: non c'e' una cella scelta guardando il")
print(" risultato, quindi non c'e' un deflated-Sharpe da calcolare. La griglia ridotta")
print(" per budget e': i muri girano a 2.000 path invece di 4.000 (risoluzione ~1%,")
print(" dichiarata), e le tabelle non-4b sono all'ancora canonica.")
print(f"\n{clock()} fine.")
if __name__ == "__main__":
main()