Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d597fc64e8aeae75327a02a826be3c63960239a9
PythagorasGoal/scripts/research/r0822e_scale_proto.py
T

699 lines
40 KiB
Python
Raw Blame History

"""r0822e — PROTOTIPO ISOLATO della CHIAVE DI SCALA del libro live (filone SCALE-SPEC).
⚠️ QUESTO FILE NON E' PRODUZIONE E NON E' IMPORTATO DA NIENTE IN PRODUZIONE.
Non scrive nulla: ne' config/, ne' data/, ne' src/. Non manda ordini. Non tocca il conto.
La sezione (0) VERIFICA entrambe le cose a runtime invece di dichiararle:
- nessun file fuori da scripts/research/ nomina questo modulo;
- il sorgente di questo file non contiene una sola primitiva di scrittura.
Documento di specifica che accompagna questo prototipo: docs/research/SPEC-scale-key.md
COSA DIMOSTRA
Il filone WORST-DAY (r0822c) ha misurato che il gradino di leva 1,00x -> 1,25x vale
14,7 -> 11,6 anni al capitale-rendita, che C1-C5 passano, e che **C6 (processo) FALLISCE**:
`config/live.json` non ha una chiave di scala e il cap e' un CLAMP, non un moltiplicatore
(morde su 3 osservazioni-asset su 5.650). Questo prototipo mostra la MECCANICA della chiave
che manca, senza costruirla in produzione:
(1) SANITY la copia della funzione a scala 1.0 e' BIT-EXACT alla produzione
(2) DOVE la catena segnale -> target -> ordine, letta dal sorgente vero con le righe
(3) SIMMETRIA la scala e' identica sui due sleeve (e il controllo positivo la smaschera)
(4) ORDINE scala PRIMA o DOPO il clamp: i due ordini danno k_eff diversi, misurato
(5) REPLICA i numeri a k=1,00x coincidono con quelli pubblicati da WORST-DAY
(6) GUARDIE tetto composito, disaster-SL, liquidazione: quale guardia MORDE per prima
(7) ANTI-CREEP il gate mancante: la finestra recente non deve poter autorizzare il gradino
uv run python scripts/research/r0822e_scale_proto.py
"""
from __future__ import annotations
import json
import re
import subprocess
import sys
import time
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"))
import r0725_capcurve as CC # noqa: E402 book_series (lente LIVE del libro)
from src.data.downloader import load_data # noqa: E402
from src.live.book import ( # noqa: E402 SOLA LETTURA (nessuna di queste scrive)
CAP_UNKNOWN_USD, W_SKH, W_TP01, _cap, _read_watermark, book_net_target)
from src.live.shadow import WEIGHT # noqa: E402 0.5 = split BTC/ETH
from src.strategies.skyhook import SKH01_V2_DD # noqa: E402
from src.strategies.trend_portfolio import ( # noqa: E402
CANONICAL, TrendPortfolio, realized_vol, resample_1d, simple_returns)
ASSETS = ("BTC", "ETH")
ANN = 365.0
DELUCK = 0.89 # fattore d'ancora MISURATO (r0726_deluck_factor)
MM_RATE = 0.01 # maintenance margin perp lineari Deribit (valore prudente)
WIPE = 0.90
# ---- i numeri PUBBLICATI da WORST-DAY (sez. 33 di RESULTS-0822), per il confronto di replica ---
PUB = {
"worst_close": -0.0394, # peggior giorno in chiusura, lente LIVE
"worst_date": "2020-03-13",
"worst_min": -0.0733, # peggior minimo intra-giorno (lente accoppiata)
"L_struct": 0.1432, # peggior giorno POSSIBILE a k=1,00x, in chiusura
"cap_bite_n": 3, # osservazioni-asset in cui il clamp morde
"cap_bite_tot": 5650,
"drift_deluck": 0.1711, # r0822_growth_policy
"vol": 0.1137,
"sharpe_deluck": 1.51,
}
# ---- la PROPOSTA che questo prototipo specifica ------------------------------------------
SCALA_PROPOSTA = 1.25
LEVA_LORDA_MAX = 1.25 # tetto sul PRODOTTO n_asset * frac * scala (costante di CODICE)
MEZZO_EQUITY = 0.50 # C1: il peggior giorno possibile deve lasciare meta' equity
GRIGLIA_K = (1.00, 1.10, 1.25, 1.50, 2.00)
def sez(t: str) -> None:
print("\n" + "=" * 108)
print(f" {t}")
print("=" * 108)
# ==========================================================================================
# LA COPIA — l'unica differenza rispetto a src/live/book.book_net_target e' `scala`
# ==========================================================================================
def book_net_target_scalato(tp_frac: float, skh_sign: int, equity: float, cap: float,
weight: float = WEIGHT, scala: float = 1.0,
scala_prima_del_clamp: bool = False) -> float:
"""COPIA di src/live/book.book_net_target con la chiave di scala.
`scala_prima_del_clamp` esiste SOLO per misurare l'ordine sbagliato nella sezione (4):
la specifica propone l'ordine DOPO (scala * clamp(raw)), non PRIMA.
"""
raw = weight * equity * (W_TP01 * max(tp_frac, 0.0) + W_SKH * float(skh_sign))
if scala_prima_del_clamp:
raw = float(scala) * raw
return max(-cap, min(cap, raw))
clamped = max(-cap, min(cap, raw))
return float(scala) * clamped
def book_net_target_ASIMMETRICO(tp_frac: float, skh_sign: int, equity: float, cap: float,
weight: float = WEIGHT, scala: float = 1.0) -> float:
"""CONTROLLO POSITIVO (deve essere SMASCHERATO dal test di simmetria): la scala applicata
alla sola gamba TP01. E' esattamente cio' che fa `target_vol`, ed e' un TILT DI PESI."""
raw = weight * equity * (W_TP01 * float(scala) * max(tp_frac, 0.0) + W_SKH * float(skh_sign))
return float(scala) * 0.0 + max(-cap, min(cap, raw))
# ==========================================================================================
# (0) ISOLAMENTO — verificato, non dichiarato
# ==========================================================================================
def sezione_0_isolamento() -> None:
sez("(0) ISOLAMENTO — questo file non e' produzione, e la verifica e' qui dentro")
slug = Path(__file__).stem
try:
out = subprocess.run(["grep", "-rl", slug, str(ROOT / "src"), str(ROOT / "scripts"),
str(ROOT / "tests"), str(ROOT / "config")],
capture_output=True, text=True, timeout=60).stdout.split()
except Exception as e: # grep assente: si dichiara, non si finge
out = []
print(f" ⚠️ grep non disponibile ({type(e).__name__}): controllo non eseguito")
fuori = [p for p in out if not p.endswith(f"scripts/research/{slug}.py")]
print(f" file che nominano `{slug}` fuori da se stesso : {len(fuori)} {fuori}")
assert not fuori, "QUALCUNO IMPORTA IL PROTOTIPO: non e' piu' isolato"
src = Path(__file__).read_text()
# I nomi delle primitive vietate sono COMPOSTI a runtime: scriverli per esteso qui li
# farebbe trovare dal controllo stesso (primo tentativo: il file falliva su se' stesso).
vietate = tuple("".join(p) for p in (("write", "_text("), ("to_", "parquet("), ("to_", "csv("),
("json.", "dump("), ("os.", "remove"), ("shu", "til."),
("mk", "dir("), ("os.", "rename")))
scritture = [m for m in vietate if m in src]
aperture_w = re.findall(r"open\([^)]*['\"][wa]", src)
print(f" primitive di scrittura nel sorgente : {scritture or 'nessuna'}")
print(f" open() in scrittura/append : {len(aperture_w)}")
assert not scritture and not aperture_w, "il prototipo scrive: non e' piu' innocuo"
print(" -> il prototipo LEGGE il codice di produzione e non lo tocca. Il cron non lo vede.")
# ==========================================================================================
# (1) SANITY — la copia e' la produzione
# ==========================================================================================
def sezione_1_sanity() -> None:
sez("(1) SANITY — a scala 1,00 la copia e' BIT-EXACT alla funzione di produzione")
dif = 0.0
n = 0
for eq in (600.0, 2_000.0, 6_047.0, 50_000.0):
for cap in (300.0, eq * 0.5, eq * 0.25, 1e9):
for tp in (0.0, 0.13, 1 / 3, 0.5, 2 / 3, 0.87, 1.0, 1.4, 2.0, -1.0):
for sg in (-1, 0, 1):
a = book_net_target(tp, sg, eq, cap)
b = book_net_target_scalato(tp, sg, eq, cap, scala=1.0)
dif = max(dif, abs(a - b))
n += 1
print(f" {n} combinazioni (equity x cap x tp_frac x segno) · max|copia - produzione| = {dif:.2e}")
assert dif == 0.0, "la copia NON e' la produzione: ogni numero sotto e' senza valore"
print(" -> tutto cio' che segue e' una proprieta' della FUNZIONE DI PRODUZIONE, non di una sua imitazione.")
# ==========================================================================================
# (2) DOVE VA LA CHIAVE — la catena letta dal sorgente vero
# ==========================================================================================
def _righe(path: Path, pattern: str) -> list[tuple[int, str]]:
out = []
for i, ln in enumerate(path.read_text().splitlines(), 1):
if re.search(pattern, ln):
out.append((i, ln.rstrip()))
return out
def sezione_2_dove() -> None:
sez("(2) DOVE — la catena segnale -> target -> ordine, letta da src/live/ con le righe vere")
book = ROOT / "src" / "live" / "book.py"
shadow = ROOT / "src" / "live" / "shadow.py"
exe = ROOT / "scripts" / "live" / "book_execute.py"
for titolo, path, pat in (
("segnale TP01 (frazione long-flat, causale)", shadow, r"current_target"),
("segnale SKH01 (solo il SEGNO: nessuna taglia)", book, r"_skh_sign|_skyhook_positions\(load5m"),
("cap per-asset (clamp, watermark-bound)", book, r"^def _cap|cap = _cap\("),
("TARGET NETTO <== QUI VA LA SCALA", book, r"^def book_net_target|net = book_net_target\("),
("ordine (delta verso il target)", book, r"^def build_book_order|order = build_book_order\("),
("invio (porta il conto al target)", exe, r"rebalance_signed|ensure_disaster_sl\(inst"),
):
print(f"\n {titolo}")
for i, ln in _righe(path, pat)[:3]:
print(f" {path.relative_to(ROOT)}:{i:>4} {ln.strip()[:92]}")
print(f"""
LETTURA. `book_net_target` e' l'UNICO punto in cui le due gambe si incontrano con una
TAGLIA: sopra di lei TP01 da' una frazione e SKH01 da' un SEGNO (+1/-1/0, nessuna
grandezza); sotto di lei `build_book_order` calcola solo un DELTA verso il target.
Percio' e' l'unico punto in cui un moltiplicatore e' un puro cambio di scala:
· su TP01 (`target_vol`) -> scala il 75% del libro e lascia fermo il 25% = TILT DI PESI
· su SKH01 -> non c'e' niente da scalare: e' un segno
· su `WEIGHT` (0.5) -> e' il SPLIT BTC/ETH, ed e' condiviso con shadow_report/dashboard
(percorso TP01-only): cambiarlo cambierebbe un ALTRO libro
· su `W_TP01`/`W_SKH` -> sono il RAPPORTO 75/25 e sono legati per test ai pesi del
backtest (`deribit_book_sleeves`): moltiplicarli entrambi
romperebbe quella parita' e confonderebbe scala e mix
· su `build_book_order` -> scalerebbe il DELTA, non il target: cambia la VELOCITA' di
convergenza, non la taglia, e il reconciler litigherebbe con se'
· dentro l'esecutore -> il report e il log stamperebbero un target diverso da quello
eseguito (il difetto che questo progetto ha gia' pagato 3 volte)""")
# ==========================================================================================
# (3) SIMMETRIA
# ==========================================================================================
def sezione_3_simmetria() -> None:
sez("(3) SIMMETRIA — la scala e' identica sui due sleeve, e il controllo positivo lo prova")
eq, cap = 6_047.0, 1e12 # cap disattivato: qui si guarda SOLO la formula
print(f" {'k':>6}{'solo TP01 (tp=1,sg=0)':>24}{'solo SKH01 (tp=0,sg=1)':>25}"
f"{'rapporto TP/SKH':>18}{'peso implicito TP01':>21}")
base_ratio = None
for k in GRIGLIA_K:
t = book_net_target_scalato(1.0, 0, eq, cap, scala=k)
s = book_net_target_scalato(0.0, 1, eq, cap, scala=k)
rap = t / s
w = t / (t + s)
base_ratio = rap if base_ratio is None else base_ratio
print(f" {k:>6.2f}{t:>24,.2f}{s:>25,.2f}{rap:>18.6f}{w:>21.6f}")
assert abs(rap - W_TP01 / W_SKH) < 1e-12, "la scala ha cambiato il MIX: e' un tilt"
print(f" -> rapporto costante = W_TP01/W_SKH = {W_TP01/W_SKH:.4f} a OGNI k. La scala non e' un tilt.")
print("\n proprieta' forte (quella che il test deve asserire): target(k) == k * target(1)")
peggio = 0.0
for k in GRIGLIA_K:
for tp in (0.0, 1 / 3, 0.5, 2 / 3, 1.0, 1.35):
for sg in (-1, 0, 1):
a = book_net_target_scalato(tp, sg, eq, cap, scala=k)
b = k * book_net_target_scalato(tp, sg, eq, cap, scala=1.0)
peggio = max(peggio, abs(a - b))
print(f" max|target(k) - k*target(1)| su {len(GRIGLIA_K)*6*3} combinazioni (cap disattivato) = {peggio:.2e}")
assert peggio == 0.0
print("\n CONTROLLO POSITIVO — la stessa proprieta' su un'implementazione ASIMMETRICA")
print(" (scala sulla sola gamba TP01 = cio' che farebbe `target_vol`):")
print(f" {'k':>6}{'solo TP01':>16}{'solo SKH01':>14}{'rapporto':>12}{'peso TP01':>12} esito del test")
smascherato = False
for k in (1.00, 1.25, 1.50):
t = book_net_target_ASIMMETRICO(1.0, 0, eq, cap, scala=k)
s = book_net_target_ASIMMETRICO(0.0, 1, eq, cap, scala=k)
ok = abs(t / s - W_TP01 / W_SKH) < 1e-12
smascherato |= (not ok)
print(f" {k:>6.2f}{t:>16,.0f}{s:>14,.0f}{t/s:>12.4f}{t/(t+s):>12.6f} "
f"{'passa (k=1: non c e nulla da vedere)' if ok else 'FALLISCE <- smascherato'}")
assert smascherato, "il test di simmetria NON smaschera l'asimmetria: non sta controllando niente"
print(" -> a k=1 le due implementazioni COINCIDONO: un test che gira solo a k=1 ha potenza ZERO.")
print(" Il test di simmetria deve girare su una GRIGLIA di k, o non e' un test.")
# ==========================================================================================
# dati storici: peso TENUTO da TP01 per asset (stessa costruzione di r0822c gambe_tp01)
# ==========================================================================================
def pesi_tp01() -> dict:
out = {}
tv = CANONICAL["target_vol"]
for a in ASSETS:
df = resample_1d(load_data(a, "1h"))
c = df["close"].values.astype(float)
r = simple_returns(c)
tgt = np.nan_to_num(np.asarray(TrendPortfolio(**CANONICAL).target_series(df), float))
held = np.zeros(len(tgt))
held[1:] = tgt[:-1] # cio' che il libro TIENE il giorno d
vol = realized_vol(r, CANONICAL["vol_win_days"], 365.25)
idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
out[a] = pd.DataFrame({"held": held, "ret": r, "vol_prev": np.r_[np.nan, vol[:-1]]},
index=idx).iloc[1:]
out[a] = out[a].assign(z=out[a]["ret"] / (out[a]["vol_prev"] / np.sqrt(365.25)))
out[a].attrs["tv"] = tv
return out
# ==========================================================================================
# (4) L'ORDINE COL CAP
# ==========================================================================================
def sezione_4_ordine(TPQ: dict, frac: float) -> None:
sez("(4) L'ORDINE COL CAP — il punto sottile: scala PRIMA o DOPO il clamp")
print(f"""
Il cap e' un CLAMP: net = clamp(0.5*E*(0.75*tp + 0.25*sg), ±cap), cap = frac*E = {frac:.2f}*E.
A piena convinzione (tp=1, sg=+1) il grezzo vale ESATTAMENTE 0.5*E*1.0 = {0.5:.2f}*E = cap.
Quindi il clamp e' *tangente* al libro proprio nei giorni di massima esposizione.""")
E = 1.0 # tutto in frazione di equity
cap = frac * E
print(f"\n {'k':>6}{'ordine':>10}{'giorni-asset':>14}{'clampati':>11}{'quota':>9}"
f"{'k_eff medio':>13}{'k_eff a piena convinz.':>24}{'lordo max (x E)':>17}")
for k in GRIGLIA_K:
for prima in (True, False):
tot = n_cl = 0
num = den = 0.0
for a in ASSETS:
h = np.clip(TPQ[a]["held"].values, 0.0, None)
for sg in (1,): # SKH long: il caso che spinge di piu'
base = np.array([book_net_target_scalato(x, sg, E, cap, scala=1.0) for x in h])
got = np.array([book_net_target_scalato(x, sg, E, cap, scala=k,
scala_prima_del_clamp=prima) for x in h])
m = np.abs(base) > 1e-12
num += float(np.sum(got[m]))
den += float(np.sum(base[m]))
n_cl += int(np.sum(np.abs(got) < np.abs(k * base) - 1e-12))
tot += int(m.sum())
piena = book_net_target_scalato(1.0, 1, E, cap, scala=k, scala_prima_del_clamp=prima) / \
book_net_target_scalato(1.0, 1, E, cap, scala=1.0)
lordo = len(ASSETS) * abs(book_net_target_scalato(1.0, 1, E, cap, scala=k,
scala_prima_del_clamp=prima))
print(f" {k:>6.2f}{'PRIMA' if prima else 'DOPO':>10}{tot:>14}{n_cl:>11}{n_cl/tot:>9.2%}"
f"{num/den:>13.4f}{piena:>24.4f}{lordo:>17.3f}")
print(f"""
LETTURA — l'ordine non e' un dettaglio, e' la differenza fra una scala e un'altra strategia.
· SCALA PRIMA DEL CLAMP: il clamp la mangia proprio nei giorni di massima convinzione
(k_eff a piena convinzione = 1,00 a OGNI k). Il risultato non e' "il libro a leva k":
e' un libro che alza i giorni piccoli e lascia fermi i grandi, cioe' un APPIATTIMENTO
del profilo di convinzione. E' un cambio di FORMA travestito da cambio di taglia — la
curva g(k) di growth_policy/WORST-DAY NON descrive quel libro, perche' quella curva e'
calcolata su k * r, cioe' sull'ordine DOPO.
· SCALA DOPO IL CLAMP: k_eff = k esattamente, in ogni giorno e in ogni stato del segnale.
E' l'unico ordine per cui "leva k" significa la stessa cosa nel codice e nel backtest.
⚠️ IL PREZZO, e va detto perche' e' il cuore della specifica: con l'ordine DOPO il cap
SMETTE di essere il tetto assoluto del nozionale. Il lordo massimo diventa
n_asset * frac * k = {len(ASSETS)} * {frac:.2f} * k, cioe' {len(ASSETS)*frac*SCALA_PROPOSTA:.3f}x l'equity a k={SCALA_PROPOSTA:.2f}.
E' ESATTAMENTE la quantita' che `test_leva_massima_da_config_resta_sotto_o_uguale_a_1x`
(tests/test_fee_sensitivity.py) asserisce essere <= 1x. Quel test DEVE rompersi: sta
misurando `frac * n_asset` mentre la grandezza vera diventa `frac * n_asset * scala`.""")
# ==========================================================================================
# (5) REPLICA DEI NUMERI DI WORST-DAY a k = 1,00
# ==========================================================================================
def sezione_5_replica(TPQ: dict, frac: float) -> dict:
sez("(5) REPLICA — a k=1,00 i numeri devono coincidere con quelli pubblicati da WORST-DAY")
book = CC.book_series("hourly")
r = book.values.astype(float)
r_del = r - (1.0 - DELUCK) * float(r.mean())
m = {
"n": len(r),
"drift_deluck": float(r_del.mean()) * ANN,
"vol": float(r_del.std(ddof=0)) * np.sqrt(ANN),
"sharpe_deluck": float(r_del.mean() / r_del.std(ddof=0)) * np.sqrt(ANN),
"worst_close": float(r_del.min()),
"worst_date": str(book.index[int(np.argmin(r_del))].date()),
}
print(f"\n serie del libro LIVE (`book_series('hourly')`), {m['n']} giorni "
f"{book.index.min().date()} -> {book.index.max().date()}")
print(f" {'grandezza':>26}{'qui':>12}{'pubblicato':>13}{'scarto':>10}")
for nome, ch, pu, fmt in (("drift de-luckato x0.89", "drift_deluck", PUB["drift_deluck"], "%"),
("vol annua", "vol", PUB["vol"], "%"),
("Sharpe de-luckato", "sharpe_deluck", PUB["sharpe_deluck"], "f"),
("peggior giorno (chiusura)", "worst_close", PUB["worst_close"], "%")):
v = m[ch]
s = f"{v:>12.2%}{pu:>13.2%}{v-pu:>10.2%}" if fmt == "%" else f"{v:>12.3f}{pu:>13.3f}{v-pu:>10.3f}"
print(f" {nome:>26}{s}")
print(f" {'data del peggior giorno':>26}{m['worst_date']:>12}{PUB['worst_date']:>13}"
f"{' coincide' if m['worst_date'] == PUB['worst_date'] else ' DIVERSA':>10}")
assert m["worst_date"] == PUB["worst_date"]
assert abs(m["worst_close"] - PUB["worst_close"]) < 5e-4
# --- il clamp morde su 3 osservazioni-asset su 5.650
n_cap = sum(int(((0.5 * W_TP01 * TPQ[a]["held"].values + 0.5 * W_SKH) > frac).sum()) for a in ASSETS)
n_tot = sum(len(TPQ[a]) for a in ASSETS)
print(f"\n il CLAMP a frac={frac:.2f} morde su {n_cap} osservazioni-asset su {n_tot} "
f"({n_cap/n_tot:.3%}) [pubblicato: {PUB['cap_bite_n']}/{PUB['cap_bite_tot']}]")
for f2 in (0.50, 0.5625, 0.625, 0.75):
nn = sum(int(((0.5 * W_TP01 * TPQ[a]["held"].values + 0.5 * W_SKH) > f2).sum()) for a in ASSETS)
print(f" frac={f2:.4f} -> morde {nn:>3} volte · leva lorda MASSIMA {2*f2:.3f}x · "
f"leva EFFETTIVA mediana {np.median(np.concatenate([np.minimum(0.5*W_TP01*TPQ[a]['held'].values+0.5*W_SKH, f2)*2 for a in ASSETS])):.3f}x")
print(" -> alzare `frac` NON alza la leva: alza un tetto che non tocca quasi mai il libro.")
# --- il bound strutturale a k = 1
tv = CANONICAL["target_vol"]
sl_l = SKH01_V2_DD.sl_pct
ex_giu = {}
for a in ASSETS:
d = load_data(a, "1h")
ex_giu[a] = float((d["close"].values / d["open"].values - 1.0).min())
lim_tp, lim_sk = {}, {}
for a in ASSETS:
d = TPQ[a].dropna(subset=["z"])
dv = d[d["vol_prev"] >= tv]
z = float(dv["z"].min())
lim_tp[a] = 0.5 * W_TP01 * tv * abs(z) / np.sqrt(365.25)
lim_sk[a] = 0.5 * W_SKH * (sl_l + abs(ex_giu[a]))
L = sum(lim_tp[a] + lim_sk[a] for a in ASSETS)
print(f"\n bound STRUTTURALE (scenario crash: TP01 a piena convinzione + SKH01 long stoppato)")
for a in ASSETS:
print(f" {a}: TP01 {-lim_tp[a]:>7.2%} SKH01 {-lim_sk[a]:>7.2%} -> {-(lim_tp[a]+lim_sk[a]):>7.2%}")
print(f" TOTALE a k=1,00x {-L:>7.2%} [pubblicato: {-PUB['L_struct']:.2%}]")
assert abs(L - PUB["L_struct"]) < 5e-3, "il bound strutturale non si replica"
print(" -> replica superata: la sezione (6) puo' scalare QUESTI numeri e non dei numeri copiati.")
return {"r_del": r_del, "L": L, "book": book}
# ==========================================================================================
# (6) LE GUARDIE — quale morde per prima
# ==========================================================================================
def sezione_6_guardie(rep: dict, TPQ: dict, frac: float, sl_pct: float) -> None:
sez("(6) GUARDIE — il tetto e' sul PRODOTTO, e la guardia che morde per prima non e' quella attesa")
L = rep["L"]
g_max = sum(0.5 * W_TP01 * TPQ[a]["held"].max() + 0.5 * W_SKH for a in ASSETS)
m_min = PUB["worst_min"] # peggior minimo intra-giorno misurato (accoppiato)
print(f"""
Tre grandezze scalano LINEARMENTE con k e vanno lette insieme (sl disaster = {sl_pct:.0%}):
A) peggior giorno POSSIBILE in chiusura : k * {L:.2%}
B) costo in equity se scatta il disaster-SL : k * n_asset * frac * {sl_pct:.0%}
(il -{sl_pct:.0%} e' un movimento di PREZZO sul mark; l'impatto in equity e' quel
movimento moltiplicato per il LORDO, quindi cresce con k anche se sl_pct non cambia)
C) distanza dalla liquidazione sul MINIMO : margine = x_liq / (|min| * k)
""")
print(f" {'k':>6}{'lordo max':>11}{'A) peggior giorno':>19}{'B) costo disaster-SL':>22}"
f"{'C) x_liq':>10}{'min intra k':>13}{'margine C':>11} verdetto")
for k in GRIGLIA_K:
G = min(g_max * k, len(ASSETS) * frac * k)
x_liq = (1 - MM_RATE * G) / (G * (1 - MM_RATE))
A = k * L
B = k * len(ASSETS) * frac * sl_pct
mk = abs(m_min) * k
margine = x_liq / mk
v = []
if A > MEZZO_EQUITY:
v.append("A sfonda meta' equity")
if B > MEZZO_EQUITY:
v.append("B sfonda meta' equity")
if margine < 3.0:
v.append("C sotto 3x")
if k * len(ASSETS) * frac > LEVA_LORDA_MAX:
v.append(f"oltre il tetto {LEVA_LORDA_MAX:.2f}x")
print(f" {k:>6.2f}{G:>11.3f}{A:>19.2%}{B:>22.2%}{x_liq:>10.1%}{mk:>13.2%}"
f"{margine:>11.1f}x {'; '.join(v) if v else 'ok'}")
k_da_A = MEZZO_EQUITY / L
k_da_B = MEZZO_EQUITY / (len(ASSETS) * frac * sl_pct)
print(f"""
📌 IL RISULTATO CHE WORST-DAY NON AVEVA CALCOLATO: la guardia che morde per prima NON e'
il peggior giorno possibile, e' il DISASTER-SL.
· dal peggior giorno (A): k <= {MEZZO_EQUITY:.0%} / {L:.2%} = {k_da_A:.2f}x
· dal disaster-SL (B): k <= {MEZZO_EQUITY:.0%} / ({len(ASSETS)} * {frac:.2f} * {sl_pct:.0%}) = {k_da_B:.2f}x
Il disaster-SL e' {k_da_A/k_da_B:.1f}x piu' stringente. A k={SCALA_PROPOSTA:.2f} il margine di (A) e'
{MEZZO_EQUITY/(SCALA_PROPOSTA*L):.1f}x mentre quello di (B) e' solo {MEZZO_EQUITY/(SCALA_PROPOSTA*len(ASSETS)*frac*sl_pct):.2f}x.
-> il tetto di codice non puo' essere un numero solo: deve essere l'INVARIANTE
`n_asset * frac * scala * disaster_sl_pct <= {MEZZO_EQUITY:.2f}`, che lega TRE chiavi.
Con sl_pct={sl_pct:.0%} e frac={frac:.2f} quell'invariante da' scala <= {k_da_B:.2f}x.
⚠️ E il buco che un tetto sulla SOLA chiave lascerebbe aperto: `frac` e `scala` MOLTIPLICANO.
Un tetto "scala <= {SCALA_PROPOSTA:.2f}" e' rispettato anche da frac=0.75 + scala={SCALA_PROPOSTA:.2f}
= leva lorda {2*0.75*SCALA_PROPOSTA:.3f}x. Il tetto va messo sul PRODOTTO.""")
print(f"\n {'frac':>8}{'scala':>8}{'prodotto = leva lorda':>24}{'tetto scala-only':>18}{'tetto sul PRODOTTO':>21}")
for f2, k in ((0.50, 1.00), (0.50, 1.25), (0.50, 1.50), (0.625, 1.25), (0.75, 1.25), (0.50, 2.00)):
p = len(ASSETS) * f2 * k
print(f" {f2:>8.3f}{k:>8.2f}{p:>24.3f}{'passa' if k <= SCALA_PROPOSTA else 'BLOCCA':>18}"
f"{'passa' if p <= LEVA_LORDA_MAX else 'BLOCCA':>21}")
# ==========================================================================================
# (6-bis) GLI STATI DELL'EQUITY — la scala non deve rendere il FALLBACK piu' permissivo
# ==========================================================================================
def cap_copia(equity, real_equity, eq_fallback, fixed, frac, wm):
"""COPIA di src/live/book._cap (verificata sotto contro l'originale). Riprodotta qui perche'
l'originale legge il watermark REALE da disco e questo prototipo non deve dipenderne."""
if frac is None:
return fixed
trusted = (real_equity is not None) and (not eq_fallback) and (equity is not None) and (equity > 0)
if trusted:
return float(equity) * float(frac)
return min(fixed, wm * float(frac)) if wm is not None else min(fixed, CAP_UNKNOWN_USD)
def sezione_6bis_equity(frac: float, fixed: float) -> None:
sez("(6-bis) GLI STATI DELL'EQUITY — la regola proposta: la scala vive SOLO sul percorso fidato")
wm = _read_watermark()
print(f" watermark reale sul disco (letto, non scritto): "
f"{'$%.2f' % wm if wm is not None else 'assente'}")
# la copia e' la produzione, verificata sui due rami che non dipendono dal file
d = max(abs(cap_copia(6047.0, 6047.0, None, fixed, frac, wm) - _cap(6047.0, 6047.0, None)),
abs(cap_copia(2000.0, None, "fallback", fixed, frac, wm) - _cap(2000.0, None, "fallback")))
print(f" max|copia del cap - produzione| sui due rami reali = {d:.2e}")
assert d == 0.0, "la copia del cap non e' la produzione"
E_VERO = wm if wm is not None else 635.0
print(f"""
Proposta: `scala_effettiva = scala se l'equity REALE e' leggibile, altrimenti 1.00`.
Motivo: una scala e' una decisione di rischio presa CONOSCENDO il conto; quando il conto
non si sa quanto vale, non si prende. Cosi' il vincolo "il fallback non e' piu' permissivo"
e' vero PER COSTRUZIONE e non per aritmetica.
Colonna che conta: LORDO / equity VERA (${E_VERO:,.2f}), cioe' la leva davvero corsa.""")
scen = [
("normale (equity reale letta)", E_VERO, E_VERO, None, wm),
("online ma equity illeggibile", 2000.0, None, "fallback", wm),
("watermark assente (1o avvio)", 2000.0, None, "fallback", None),
("watermark vecchio e basso", 2000.0, None, "fallback", 100.0),
("dopo un deposito a $6.047", 6047.0, 6047.0, None, 6047.0),
]
print(f"\n {'stato':>32}{'cap':>11}{'scala eff.':>12}{'lordo max':>12}"
f"{'/ equity vera':>15}{'tetto rispettato':>18}")
for nome, eq, real, fb, w in scen:
cap = cap_copia(eq, real, fb, fixed, frac, w)
k_eff = SCALA_PROPOSTA if (real is not None and not fb) else 1.0
lordo = len(ASSETS) * cap * k_eff
# l'equity VERA e' quella del conto (nota: qui la si conosce), NON il watermark: un
# watermark vecchio e basso non rende il conto piccolo, rende il libro SOTTO-esposto.
ev = real if real is not None else E_VERO
lev = lordo / ev
print(f" {nome:>32}{cap:>11,.2f}{k_eff:>12.2f}{lordo:>12,.2f}{lev:>15.3f}x"
f"{'si' if lev <= LEVA_LORDA_MAX + 1e-9 else 'NO':>18}")
print("""
-> in nessuno stato degradato la leva supera quella dello stato normale: il fallback resta
la configurazione PIU' PRUDENTE, come oggi. E il caso "watermark assente" resta a
CAP_UNKNOWN_USD ($300) con scala 1.00: due protezioni indipendenti, non una sola.
⚠️ PREZZO DICHIARATO: nei giri in cui l'equity non e' leggibile il libro gira a k=1 e al
giro dopo torna a k -> un ribilancio in piu' per ogni oscillazione. Costo: un delta di
(k-1)*frac*E per asset, cioe' ~$%.2f a testa sul conto di oggi, a 3,5 bps = $%.4f.
Frequenza MISURATA sui log (`logs/cron_book.log`): %s.""" % (
(SCALA_PROPOSTA - 1) * frac * E_VERO,
(SCALA_PROPOSTA - 1) * frac * E_VERO * 0.00035 * 2,
_freq_fallback()))
def _freq_fallback() -> str:
"""Quante volte il ramo 'equity fallback' e' stato preso davvero. Un prezzo dichiarato senza
la sua frequenza non e' un prezzo."""
log = ROOT / "logs" / "cron_book.log"
if not log.exists():
return "log assente, NON misurata"
try:
txt = log.read_text(errors="ignore")
except Exception:
return "log illeggibile, NON misurata"
giri = txt.count("BOOK DERIBIT LIVE EXECUTE")
fb = txt.count("EQUITY FALLBACK")
off = txt.count("conto non leggibile")
return (f"{fb} giri su {giri} ({fb/max(giri,1):.2%}); "
f"{off} giri sono invece OFFLINE (nessun ordine, la scala non si pone)")
# ==========================================================================================
# (7) IL GATE MANCANTE — anti-creep, e la finestra recente non deve poter autorizzare
# ==========================================================================================
def g_exact(r: np.ndarray, k: float) -> float:
x = 1.0 + k * r
if np.any(x <= 0):
return -np.inf
return float(np.expm1(ANN * np.mean(np.log(x))))
def k_star(r: np.ndarray, kmax: float = 30.0, step: float = 0.05) -> float:
ks = np.arange(step, kmax + step, step)
return float(ks[int(np.nanargmax([g_exact(r, k) for k in ks]))])
def _k_ammesso(r: np.ndarray, L: float) -> dict:
"""k autorizzato dal criterio: mezzo-Kelly sul drift a -2 SE, con la coda TRONCATA al bound
strutturale (senza troncatura E[log(1+kR)] non esiste). Deliberatamente SEMPLIFICATO
rispetto a r0822c (nessuna GPD): qui interessa la MECCANICA del gate, non il suo numero."""
mu, sd = float(r.mean()), float(r.std(ddof=0))
se = sd * np.sqrt(ANN) / np.sqrt(len(r) / ANN)
r2 = np.maximum(r - mu + (mu * ANN - 2 * se) / ANN, -L) # -2SE + coda troncata
ks = k_star(r2)
return {"drift_pt": mu * ANN, "se": se, "drift_2se": mu * ANN - 2 * se,
"k_star": ks, "k_ammesso": 0.5 * ks, "n": len(r)}
def sezione_7_gate(rep: dict) -> None:
sez("(7) IL GATE MANCANTE — la scala e' invisibile allo Sharpe: serve un gate su un ALTRO asse")
r = rep["r_del"]
L = rep["L"]
book = rep["book"]
print("\n (a) PERCHE' i gate esistenti non possono servire: lo Sharpe e' INVARIANTE alla scala.")
print(f" {'k':>6}{'Sharpe':>10}{'drift':>10}{'vol':>10}{'maxDD':>10}{'g(k) esatto':>14}")
for k in GRIGLIA_K:
x = k * r
eq = np.cumprod(1.0 + x)
dd = float((eq / np.maximum.accumulate(eq) - 1.0).min())
print(f" {k:>6.2f}{x.mean()/x.std(ddof=0)*np.sqrt(ANN):>10.4f}{x.mean()*ANN:>10.2%}"
f"{x.std(ddof=0)*np.sqrt(ANN):>10.2%}{dd:>10.2%}{g_exact(r,k):>14.2%}")
print(" -> `deflated_sharpe` e `marginal_vs_tp01` leggono la 1a colonna: non FALLISCONO,")
print(" NON VEDONO la variabile. Un gate sulla scala deve guardare g(k), il DD e la rovina.")
print("\n (b) IL CRITERIO ANTI-RECENCY — «una chiave di scala si alza dopo un mese buono».")
print(" Forma verificabile: il gradino dev'essere autorizzato ANCHE dai dati che")
print(" ESCLUDONO la finestra recente che ha motivato la richiesta.")
print(f"\n {'finestra':>34}{'n':>7}{'drift punto':>13}{'drift -2SE':>12}{'k*':>8}"
f"{'k ammesso':>11} {SCALA_PROPOSTA:.2f}x?")
fin = [("TUTTA la storia", r),
("esclusi gli ultimi 90 giorni", r[:-90]),
("esclusi gli ultimi 180 giorni", r[:-180]),
("esclusi gli ultimi 365 giorni", r[:-365]),
("SOLO gli ultimi 365 giorni", r[-365:])]
for nome, rr in fin:
d = _k_ammesso(rr, L)
ok = SCALA_PROPOSTA <= d["k_ammesso"]
print(f" {nome:>34}{d['n']:>7}{d['drift_pt']:>13.2%}{d['drift_2se']:>12.2%}"
f"{d['k_star']:>8.2f}{d['k_ammesso']:>11.2f} {'SI' if ok else 'NO'}")
d_full = _k_ammesso(r, L)
d_ex = _k_ammesso(r[:-90], L)
print(f"""
⚠️ LETTURA ONESTA, e la prima stesura di questa sezione l'aveva sbagliata. La riga
«SOLO gli ultimi 365 giorni» NON e' l'esempio di «un anno buono»: quell'anno e' stato
DEBOLE (drift di punto {_k_ammesso(r[-365:], L)['drift_pt']:.2%}). Cio' che quella riga mostra e' solo che su
365 osservazioni l'errore standard divora tutto — vero e insufficiente, perche' il rischio
da coprire e' l'opposto: un tratto recente BUONO che alza la richiesta.""")
print("\n Percio' la domanda va posta al contrario: QUANTO puo' spostare il k ammesso un")
print(" tratto recente artificialmente buono? Si sovrascrive il drift degli ultimi 90")
print(" giorni e si guarda quanto si muove il criterio, sull'INTERA storia.")
print(f"\n {'ultimi 90g portati a':>24}{'drift punto (full)':>21}{'drift -2SE':>13}"
f"{'k ammesso':>11}{'delta vs base':>15}")
base_k = d_full["k_ammesso"]
for shock in (0.0, 0.25, 0.50, 1.00, 2.00):
rr = r.copy()
rr[-90:] = rr[-90:] - rr[-90:].mean() + shock / ANN
d = _k_ammesso(rr, L)
print(f" {shock:>23.0%}{d['drift_pt']:>21.2%}{d['drift_2se']:>13.2%}"
f"{d['k_ammesso']:>11.2f}{d['k_ammesso']-base_k:>+15.2f}")
print(f"""
-> un trimestre portato a +100%/anno (assurdo) muove il k ammesso di poche unita' su una
base di {base_k:.2f}: e' il -2 SE che fa il lavoro, perche' un tratto eccezionale alza
insieme il drift di punto E la sua incertezza. **Ma questo NON basta come gate**: il
criterio resta comunque una funzione dei dati recenti, e la protezione vera e' la
DOPPIA VALUTAZIONE — la richiesta dev'essere autorizzata anche da `r[:-90]`
(qui k ammesso {d_ex['k_ammesso']:.2f} contro {base_k:.2f}), cioe' da dati che NON contengono
la finestra che ha motivato la richiesta. E' l'analogo, sull'asse della scala, di
`select_cell_insample`.
⚠️ E va detto il numero che rende tutto questo poco drammatico OGGI: con questa versione
semplificata il k ammesso e' {base_k:.2f}x, cioe' il gradino di 1,25x non e' vicino a
quel vincolo. **Il vincolo che morde e' quello della sez.(6) — il disaster-SL — non
il mezzo-Kelly.** Il criterio anti-recency serve al gradino SUCCESSIVO, non a questo.""")
print("\n (c) LA CADENZA — un gate che si valuta solo quando lo si chiede non protegge da niente.")
ultimo = book.index.max()
print(f" ultimo giorno della serie: {ultimo.date()}. Il sorvegliante deve rispondere a")
print(" TRE domande diverse, e sono davvero tre (azioni diverse):")
print(" 1. la config dichiara una scala che il giornale non ha mai autorizzato? -> 🚨")
print(" 2. il criterio che autorizzo' la scala CORRENTE passa ancora oggi? -> 🚨 (si scende)")
print(" 3. il prodotto frac*scala*n_asset ha superato il tetto di codice? -> 🚨 (si blocca)")
print(" Nessuna delle tre e' deducibile dalle altre due.")
# ==========================================================================================
def main() -> None:
t0 = time.time()
print("=" * 108)
print(" r0822e — PROTOTIPO ISOLATO della CHIAVE DI SCALA (filone SCALE-SPEC)")
print(" Specifica: docs/research/SPEC-scale-key.md · NON e' produzione, non scrive nulla")
print("=" * 108)
cfg = json.loads((ROOT / "config" / "live.json").read_text())
frac = float(cfg["max_notional_per_asset_frac"])
sl_pct = float(cfg["disaster_sl_pct"])
print(f"\n config/live.json (SOLA LETTURA): frac={frac} · cap fisso ${cfg['max_notional_per_asset_usd']:,.0f}"
f" · min order ${cfg['min_order_usd']} · disaster-SL {sl_pct:.0%}")
print(f" chiavi presenti: {sorted(k for k in cfg if not k.startswith('_'))}")
print(" -> nessuna chiave di scala. E' esattamente il buco che questo filone specifica.")
sezione_0_isolamento()
sezione_1_sanity()
sezione_2_dove()
sezione_3_simmetria()
TPQ = pesi_tp01()
sezione_4_ordine(TPQ, frac)
rep = sezione_5_replica(TPQ, frac)
sezione_6_guardie(rep, TPQ, frac, sl_pct)
sezione_6bis_equity(frac, float(cfg['max_notional_per_asset_usd']))
sezione_7_gate(rep)
sez("RIEPILOGO — cosa questo prototipo ha DIMOSTRATO (e cosa resta alla specifica)")
print(f"""
DIMOSTRATO QUI
· la copia e' bit-exact alla produzione a k=1 -> le proprieta' misurate sono SUE;
· la scala applicata in `book_net_target` e' simmetrica sui due sleeve a ogni k, e il
controllo positivo prova che il test ha potenza SOLO se gira su una griglia di k;
· l'ordine col clamp decide se "leva k" significa la stessa cosa nel codice e nel
backtest: DOPO il clamp k_eff = k sempre; PRIMA il clamp la mangia a piena convinzione;
· i numeri a k=1,00 replicano WORST-DAY (peggior giorno, data, bound strutturale, morso
del clamp) -> le righe scalate della sez.(6) non sono numeri copiati;
· la guardia che morde per prima e' il DISASTER-SL, non il peggior giorno: il tetto va
messo sul PRODOTTO `n_asset * frac * scala * disaster_sl_pct`;
· lo Sharpe e' invariante alla scala a tutte le celle -> nessun gate esistente la vede;
· il criterio anti-recency e' misurabile: si valuta escludendo la finestra recente.
NON DIMOSTRATO QUI, E VA DETTO
· non e' un backtest della leva: g(k) qui e' la forma chiusa su una serie storica, con
lo stesso limite gia' dichiarato da growth_policy (ritorno indipendente dalla size);
· la lente e' CLOSE-ONLY tranne dove usa il minimo pubblicato da WORST-DAY: su una regola
a UN giorno close-only e' CIECA (25/07), e la sez.(6) prende quel numero da fuori;
· il funding dei perpetual non e' in nessuna di queste serie (difetto trovato dal filone
FUNDING): a k=1 il libro paga gia' ~1,5%/anno che nessun backtest sottrae, e il gradino
lo amplifica del 25%;
· niente qui autorizza il gradino: autorizzarlo e' il gate della specifica, non un run.
tempo: {time.time()-t0:.1f}s""")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink