GATE SCALA-01: la chiave di scala esiste, ed e' INERTE (SPEC §8 punti 1-4)

Chiude il debito che CLAUDE.md dichiarava da settimane: la regola "ogni cambio
di scala passa dal cap di config, non da target_vol" NON era implementabile
perche' una chiave di scala non esisteva — qualunque cambio sarebbe finito su
WEIGHT/W_TP01/W_SKH, cioe' codice su un percorso con soldi veri e per giunta
nel posto sbagliato (W_TP01/W_SKH sono il RAPPORTO 75/25, non la taglia).

Specifica gia' scritta in docs/research/SPEC-scale-key.md (618 righe, 9
condizioni di gate, prototipo). Non ho progettato: ho eseguito i punti 1-4.

🚨 config/live.json NON E' STATO TOCCATO. La chiave e' assente, vale 1,00, e
T7 dimostra bit-exact che il libro e' quello di ieri (max|diff| = 0.0).
Verificato anche a runtime: book_execute in dry-run da' gli stessi target del
cron delle 15:47 (BTC $+355, ETH $+214).

LE QUATTRO DECISIONI CHE NON SONO DI COMODO
- La scala si applica DOPO il clamp. Prima, il cap se la mangerebbe proprio
  nei giorni di massima convinzione (a tp=1/sg=+1 il grezzo vale esattamente
  cap => k_eff tornerebbe a 1,00 a ogni k): sarebbe un cambio di FORMA
  travestito da cambio di taglia, e la curva g(k) con cui il gradino viene
  autorizzato non descriverebbe quel libro. Prezzo dichiarato: il cap diventa
  il tetto del libro UNITARIO, e la guardia sulla leva lorda va ricostruita.
- Il tetto e' sul PRODOTTO e sta nel CODICE. Sulla sola chiave lascerebbe
  aperta la porta accanto (frac 0,625 x scala 1,25 = 1,562x); in config
  sarebbe un lucchetto con la chiave attaccata. LEVA_LORDA_MAX 1,25 in
  src/live/book.py => il gradino a 1,50 richiede codice, quindi review.
- Fuori scaletta o fuori tetto = STOP, non clamp. book_execute si ferma, non
  invia, allerta (ScalaNonAutorizzata). E SCALA_LADDER (1,00 · 1,25) rende
  INESPRIMIBILE "solo un po'": 1,05 non e' prudente, e' fuori scaletta.
- La scala vive solo sul percorso fidato (equity illeggibile => 1,00), cosi'
  "il fallback non e' piu' permissivo" e' vero per costruzione. Ma la
  VALIDAZIONE avviene sempre: una config rotta non si nasconde dietro un giro
  in cui l'equity non era leggibile.

LA GUARDIA CHE MORDE PER PRIMA non e' il peggior giorno (k <= 3,49x) ma il
COSTO di un disaster-SL (k <= 1,67x, 2,1x piu' stringente): l'invariante
n_asset x frac x scala x disaster_sl_pct <= 0,50 scatta anche se qualcuno
allarga lo stop invece di alzare la scala.

TEST T1-T11 (tests/test_book_scale.py, 18 verdi). Il piu' importante e' T1b:
a k=1 l'implementazione simmetrica e quella asimmetrica danno lo STESSO
numero, quindi un test di simmetria scritto sul caso di default ha potenza
ZERO. T1b verifica che le due coincidano a k=1 (il rischio e' reale) e che
fuori da k=1 l'asserzione le SEPARI, con un'implementazione asimmetrica
scritta nel test apposta perche' fallisca.

SORVEGLIANTE scale_watch (cron_daily, 3 domande / 3 azioni / 3 stati, una
allerta per streak, marcatore scritto solo dopo invio riuscito — debito #2).
Riporta la frequenza del ramo di fallback, che sopra il 2% in 90 giorni
invaliderebbe la regola: misurata 0/1.676, coi 19 giri "paper capital"
(pre-finanziamento, dove il libro non invia) contati e dichiarati a parte.
Non puo' impedire la modifica: la rende visibile entro 24h e attribuibile.

CHIUDE il debito #5 di §5: T2/T3 sostituiscono il vecchio
test_leva_massima_da_config (che misurava frac x n_asset mentre la grandezza
vera e' frac x n_asset x scala), e T11 verifica che sia rimasto cancellato.

NON FATTO, deliberato: la chiave in config (punto 2 lo vieta), GATE SCALA-01
(A2 richiede >=30 giorni a 1,00 col sorvegliante attivo — "l'unico modo di
scoprire che il sorvegliante e' rotto mentre la leva e' ancora 1,00"),
r0726_fee_sensitivity rifatto (A7: serve solo al gradino; a 1,25x una
liquidazione costerebbe 1,25% non 1,00%, e ereditarlo sarebbe l'errore).

Nessun ordine. Suite: 825 passati.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-09-01 17:44:02 +00:00
parent 05816c49f9
commit ec8478308f
9 changed files with 944 additions and 17 deletions
+305
View File
@@ -0,0 +1,305 @@
"""T1-T11 — le guardie della CHIAVE DI SCALA (SPEC-scale-key §5).
Scritti come specifiche eseguibili: ogni test dichiara **il caso che deve farlo fallire**.
La chiave e' inerte finche' nessuno scrive un numero in config: T7 lo dimostra bit-exact.
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from src.live import book as B
from src.live import scale_watch as SW
ROOT = Path(__file__).resolve().parents[1]
GRIGLIA_K = (1.00, 1.10, 1.25, 1.50, 2.00)
GRIGLIA_TP = (0.0, 0.25, 0.461, 1.0)
GRIGLIA_SG = (-1, 0, 1)
CAP_SPENTO = 1e12 # cap disattivato: isola la scala dal clamp
EQ = 2000.0
def _cfg(**kw) -> dict:
base = dict(max_notional_per_asset_frac=0.5, disaster_sl_pct=0.3)
base.update(kw)
return base
# ---------------------------------------------------------------- T1 / T1b: la scala e' SIMMETRICA
def test_T1_la_scala_e_identica_sui_due_sleeve():
"""target(k) == k*target(1) su tutta la griglia, e il rapporto TP/SKH resta W_TP01/W_SKH.
DEVE FALLIRE SE: qualcuno applica la scala a una sola gamba (o dentro TrendPortfolio)."""
for k in GRIGLIA_K:
for tp in GRIGLIA_TP:
for sg in GRIGLIA_SG:
uno = B.book_net_target(tp, sg, EQ, CAP_SPENTO, scala=1.0)
kk = B.book_net_target(tp, sg, EQ, CAP_SPENTO, scala=k)
assert kk == pytest.approx(k * uno, abs=1e-9), (k, tp, sg)
solo_tp = B.book_net_target(1.0, 0, EQ, CAP_SPENTO, scala=k)
solo_skh = B.book_net_target(0.0, 1, EQ, CAP_SPENTO, scala=k)
assert solo_tp / solo_skh == pytest.approx(B.W_TP01 / B.W_SKH, abs=1e-12), k
def test_T1b_il_test_di_simmetria_smaschera_una_scala_asimmetrica():
"""CONTROLLO POSITIVO — il test piu' importante dell'elenco.
A k=1 l'implementazione simmetrica e quella asimmetrica danno lo STESSO numero: un test di
simmetria scritto sul solo caso di default passa sempre e non controlla niente (potenza ZERO).
Qui si verifica (a) che a k=1 le due coincidano davvero, e (b) che fuori da k=1 l'asserzione
di T1 le SEPARI.
"""
def asimmetrico(tp, sg, eq, cap, scala=1.0):
"""La scala applicata al solo TP01: e' anche un TILT di pesi, non una scala."""
raw = B.WEIGHT * eq * (B.W_TP01 * scala * max(tp, 0.0) + B.W_SKH * float(sg))
return max(-cap, min(cap, raw))
# (a) a k=1 le due implementazioni sono indistinguibili -> il test avrebbe potenza zero
for tp in GRIGLIA_TP:
for sg in GRIGLIA_SG:
assert asimmetrico(tp, sg, EQ, CAP_SPENTO, 1.0) == pytest.approx(
B.book_net_target(tp, sg, EQ, CAP_SPENTO, scala=1.0), abs=1e-12)
# (b) fuori da k=1 l'asserzione di T1 fallisce sull'implementazione asimmetrica
fallita = False
for k in (1.25, 2.00):
uno = asimmetrico(1.0, 1, EQ, CAP_SPENTO, 1.0)
kk = asimmetrico(1.0, 1, EQ, CAP_SPENTO, k)
if abs(kk - k * uno) > 1e-9:
fallita = True
peso_tp = (asimmetrico(1.0, 0, EQ, CAP_SPENTO, k)
/ (asimmetrico(1.0, 0, EQ, CAP_SPENTO, k) + asimmetrico(0.0, 1, EQ, CAP_SPENTO, k)))
assert peso_tp != pytest.approx(B.W_TP01, abs=1e-6), (
f"a k={k} il peso implicito di TP01 dovrebbe DERIVARE dallo 0,75: {peso_tp}")
assert fallita, "T1 non separa la scala asimmetrica: il controllo positivo e' rotto"
# ---------------------------------------------------------------- T2 / T3: i due tetti
def test_T2_la_leva_lorda_da_config_resta_sotto_il_tetto():
"""len(ASSETS) * frac * scala <= LEVA_LORDA_MAX, con frac e scala LETTI da config/live.json e
il tetto IMPORTATO dalla produzione.
DEVE FALLIRE SE: qualcuno alza frac, o la scala, **o entrambe di poco**."""
cfg = json.loads((ROOT / "config" / "live.json").read_text())
frac = cfg.get("max_notional_per_asset_frac")
if frac is None:
pytest.skip("cap fisso: la leva non e' derivabile da frac")
scala = float(cfg.get("book_scale_k", 1.0))
leva = len(B.ASSETS) * float(frac) * scala
assert leva <= B.LEVA_LORDA_MAX + 1e-12, (
f"leva lorda {leva:.4f}x oltre il tetto {B.LEVA_LORDA_MAX}x — il tetto e' sul PRODOTTO: "
f"n_asset {len(B.ASSETS)} x frac {frac} x scala {scala}")
def test_T3_l_invariante_disaster_sl_tiene_meta_equity():
"""n_asset * frac * scala * disaster_sl_pct <= 0.50 — la guardia che morde per PRIMA.
DEVE FALLIRE SE: qualcuno allarga disaster_sl_pct **o** alza la scala fino a 1,67x."""
cfg = json.loads((ROOT / "config" / "live.json").read_text())
frac = cfg.get("max_notional_per_asset_frac")
if frac is None:
pytest.skip("cap fisso")
costo = (len(B.ASSETS) * float(frac) * float(cfg.get("book_scale_k", 1.0))
* float(cfg["disaster_sl_pct"]))
assert costo <= B.DISASTER_SL_BUDGET + 1e-12, (
f"un episodio di disaster-SL costerebbe {100*costo:.1f}% dell'equity")
# ---------------------------------------------------------------- T4: il fallback non e' permissivo
def test_T4_il_fallback_non_e_piu_permissivo_del_percorso_fidato(tmp_path, monkeypatch):
"""In OGNI stato degradato il lordo massimo / equity vera <= quello dello stato normale.
DEVE FALLIRE SE: qualcuno applica la scala anche in fallback, o slega il cap dal watermark."""
cfg = _cfg(book_scale_k=1.25, max_notional_per_asset_usd=3000)
cpath = tmp_path / "live.json"
cpath.write_text(json.dumps(cfg))
monkeypatch.setattr(B, "CONFIG", cpath)
wm = tmp_path / "equity_seen.json"
wm.write_text(json.dumps({"real_equity": EQ}))
monkeypatch.setattr(B, "EQUITY_WATERMARK", wm)
def lordo(real_equity, eq_fallback):
cap = B._cap(equity=EQ, real_equity=real_equity, eq_fallback=eq_fallback)
sc = B._scala(real_equity=real_equity, eq_fallback=eq_fallback)
return len(B.ASSETS) * cap * sc / EQ
normale = lordo(EQ, None)
assert normale == pytest.approx(1.25, abs=1e-9)
for nome, re_, fb in (("equity illeggibile", None, None),
("eq_fallback attivo", EQ, "paper")):
assert lordo(re_, fb) <= normale + 1e-12, nome
wm.unlink() # watermark assente: due protezioni indipendenti
senza_wm = lordo(None, None)
assert senza_wm <= normale + 1e-12
assert B._scala(real_equity=None, eq_fallback=None) == 1.0
# ---------------------------------------------------------------- T5: il cap limita ancora
def test_T5_il_cap_continua_a_limitare_cio_che_limitava():
"""Il target non supera mai cap*scala, e il clamp MORDE ancora.
DEVE FALLIRE SE: qualcuno inverte l'ordine, o toglie il clamp 'tanto c'e' la scala'."""
cap = 500.0
morso = 0
for k in GRIGLIA_K:
for tp in GRIGLIA_TP:
for sg in GRIGLIA_SG:
net = B.book_net_target(tp, sg, EQ, cap, scala=k)
assert abs(net) <= cap * k + 1e-9, (k, tp, sg, net)
grezzo = B.WEIGHT * EQ * (B.W_TP01 * max(tp, 0.0) + B.W_SKH * sg)
if abs(grezzo) > cap:
morso += 1
assert abs(net) == pytest.approx(cap * k, abs=1e-9)
assert morso > 0, "il clamp non morde su nessuna osservazione: il test non controlla nulla"
# ---------------------------------------------------------------- T6: fuori tetto = STOP, non clamp
@pytest.mark.parametrize("valore,perche", [
(1.05, "fuori SCALA_LADDER: 'solo un po'' dev'essere inesprimibile"),
(1.50, "fuori SCALA_LADDER e oltre il tetto"),
(2.00, "oltre il tetto"),
])
def test_T6_una_scala_fuori_tetto_ferma_e_non_viene_tagliata(valore, perche):
"""DEVE FALLIRE SE: qualcuno 'rende robusto' il lettore con un min(scala, TETTO/...)."""
with pytest.raises(B.ScalaNonAutorizzata):
B._scala(real_equity=EQ, eq_fallback=None, cfg=_cfg(book_scale_k=valore))
def test_T6b_il_tetto_e_sul_PRODOTTO_non_sulla_sola_chiave():
"""frac 0,625 x scala 1,25 = 1,562x: passa un tetto sulla sola chiave, DEVE essere bloccato."""
with pytest.raises(B.ScalaNonAutorizzata):
B._scala(real_equity=EQ, eq_fallback=None,
cfg=_cfg(book_scale_k=1.25, max_notional_per_asset_frac=0.625))
# e la validazione avviene ANCHE sul percorso degradato: una config rotta non si nasconde
with pytest.raises(B.ScalaNonAutorizzata):
B._scala(real_equity=None, eq_fallback="paper",
cfg=_cfg(book_scale_k=1.25, max_notional_per_asset_frac=0.625))
def test_T6c_l_invariante_disaster_sl_blocca_chi_allarga_lo_stop():
"""Con sl 0,45 e scala 1,25 il costo di un disaster-SL supera mezza equity: STOP."""
with pytest.raises(B.ScalaNonAutorizzata):
B._scala(real_equity=EQ, eq_fallback=None,
cfg=_cfg(book_scale_k=1.25, disaster_sl_pct=0.45))
# ---------------------------------------------------------------- T7: inerzia bit-exact
def test_T7_senza_la_chiave_il_libro_e_quello_di_oggi():
"""Con chiave assente o = 1.0 il target e' BIT-EXACT alla formula pre-modifica.
DEVE FALLIRE SE: la modifica cambia qualcosa PRIMA che qualcuno decida di cambiarlo."""
def prima(tp, sg, eq, cap, weight=B.WEIGHT):
raw = weight * eq * (B.W_TP01 * max(tp, 0.0) + B.W_SKH * float(sg))
return max(-cap, min(cap, raw))
peggiore = 0.0
for cap in (50.0, 500.0, CAP_SPENTO):
for tp in GRIGLIA_TP:
for sg in GRIGLIA_SG:
a = prima(tp, sg, EQ, cap)
for scala in (1.0,):
peggiore = max(peggiore, abs(B.book_net_target(tp, sg, EQ, cap, scala=scala) - a))
assert peggiore == 0.0, f"la modifica non e' inerte: max|diff| = {peggiore}"
assert B._scala(real_equity=EQ, eq_fallback=None, cfg=_cfg()) == 1.0
assert float(json.loads((ROOT / "config" / "live.json").read_text())
.get("book_scale_k", 1.0)) == 1.0, "la chiave in config non e' 1,00"
# ---------------------------------------------------------------- T8: il report la dichiara
def test_T8_il_report_e_il_log_dichiarano_la_scala():
"""DEVE FALLIRE SE: un parametro che muove il nozionale resta invisibile nel report."""
src = (ROOT / "src" / "live" / "book.py").read_text()
assert "scala=scala," in src, "book_report non espone `scala` nel dict di ritorno"
ex = (ROOT / "scripts" / "live" / "book_execute.py").read_text()
assert "scala libro" in ex, "book_execute non stampa la scala"
assert "ScalaNonAutorizzata" in ex, "book_execute non gestisce la scala non autorizzata"
# ---------------------------------------------------------------- T9: derivata, non ridichiarata
def test_T9_la_scala_e_derivata_dal_codice_non_ridichiarata():
"""Il sorvegliante IMPORTA i bersagli dalla produzione. DEVE FALLIRE SE: due liste in due file."""
assert SW.LEVA_LORDA_MAX is B.LEVA_LORDA_MAX
assert SW.SCALA_LADDER is B.SCALA_LADDER
assert SW.ASSETS is B.ASSETS
assert SW.DISASTER_SL_BUDGET is B.DISASTER_SL_BUDGET
testo = (ROOT / "src" / "live" / "scale_watch.py").read_text()
for vietato in ("LEVA_LORDA_MAX =", "SCALA_LADDER =", "DISASTER_SL_BUDGET ="):
assert vietato not in testo, f"scale_watch RIDICHIARA {vietato} invece di importarlo"
# ---------------------------------------------------------------- T10: il sorvegliante segnala
def test_T10_il_sorvegliante_segnala_una_scala_non_autorizzata(tmp_path):
"""CONTROLLO POSITIVO del §6.4. DEVE FALLIRE SE: il sorvegliante e' tarato per non segnalare mai
(un rilevatore che non segnala mai e' indistinguibile da uno rotto)."""
giornale = [{"scala": 1.00, "dal": "2026-09-01T00:00:00+00:00", "gate": "iniziale"}]
muto = SW.osserva(cfg=_cfg(book_scale_k=1.00), giornale=giornale)
assert muto["stato"] == SW.OK, muto["domande"]
parlante = SW.osserva(cfg=_cfg(book_scale_k=1.25), giornale=giornale)
assert parlante["stato"] == SW.ALLARME
assert parlante["domande"]["D1"]["stato"] == SW.ALLARME, "D1 non vede la scala non autorizzata"
vuoto = SW.osserva(cfg=_cfg(book_scale_k=1.25), giornale=[])
assert vuoto["domande"]["D1"]["stato"] == SW.ALLARME, "giornale vuoto non e' un'autorizzazione"
sfora = SW.osserva(cfg=_cfg(book_scale_k=1.25, max_notional_per_asset_frac=0.625),
giornale=giornale)
assert sfora["domande"]["D3"]["stato"] == SW.ALLARME, "D3 non vede il prodotto oltre il tetto"
def test_T10b_una_allerta_per_streak_non_una_per_giro(tmp_path):
"""Disciplina degli allarmi: si dice UNA VOLTA per streak. E il marcatore 'gia' detto' si
scrive SOLO dopo un invio RIUSCITO (debito #2): con trasporto rotto si ridice."""
st = tmp_path / "state.json"
cfg = _cfg(book_scale_k=1.25)
giornale = [{"scala": 1.00, "dal": "2026-09-01T00:00:00+00:00"}]
inviati = []
def ok_sender(t):
inviati.append(t)
return True, "finto"
a = SW.run_once(state_path=st, sender=ok_sender, cfg=cfg, giornale=giornale)
b = SW.run_once(state_path=st, sender=ok_sender, cfg=cfg, giornale=giornale)
assert "D1" in a["da_dire"] and "D1" not in b["da_dire"], "ridice a ogni giro"
assert len(inviati) == 1
st.unlink()
rotti = []
def ko_sender(t):
rotti.append(t)
return False, "trasporto giu'"
SW.run_once(state_path=st, sender=ko_sender, cfg=cfg, giornale=giornale)
c = SW.run_once(state_path=st, sender=ko_sender, cfg=cfg, giornale=giornale)
assert "D1" in c["da_dire"], "un invio FALLITO ha marcato 'gia' detto': l'allarme e' perso"
assert len(rotti) == 2
def test_T10c_tre_stati_non_due(tmp_path):
"""'Non vedo' non e' 'va tutto bene': deve esistere NON MISURABILE."""
r = SW.osserva(cfg=dict(disaster_sl_pct=0.3), giornale=[]) # frac assente
assert r["domande"]["D3"]["stato"] == SW.NON_MISURABILE
fb = SW.frequenza_fallback(log=tmp_path / "non-esiste.log")
assert fb["stato"] == SW.NON_MISURABILE
# ---------------------------------------------------------------- T11: sostituito, non rilassato
def test_T11_il_vecchio_test_di_leva_e_stato_sostituito_non_rilassato():
"""DEVE FALLIRE SE: qualcuno alza la soglia del vecchio test da 1,0 a 1,25 e chiude li'."""
fs = (ROOT / "tests" / "test_fee_sensitivity.py").read_text()
assert "def test_leva_massima_da_config_resta_sotto_o_uguale_a_1x" not in fs, (
"il vecchio test misurava frac*n_asset mentre la grandezza vera e' frac*n_asset*scala: "
"va SOSTITUITO da T2/T3, non rilassato")
qui = Path(__file__).read_text()
assert "def test_T2_" in qui and "def test_T3_" in qui