merge lavoro/goldrake-u11 (secondo giro 27/09)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

scripts/cron_book.sh

@@ -35,10 +35,16 @@ mkdir -p logs
   # partire anche quando book_execute fallisce per la stessa ragione. Non blocca (exit 2
   # = allarme inviato); l'azione a un allarme e' manuale, vedi src/live/venue_watch.py.
   uv run python scripts/live/venue_watch.py --quiet || true
-  uv run python scripts/live/book_execute.py --execute
+  # LOCK SUL CONTO (#36): cuscino_watch (cron_cuscino.sh, :53) puo' mandare ordini spot USDE
+  # sullo stesso conto — separati solo dal tempo, giri fino a 659s misurati. `flock -n` (mai
+  # `-w`, un SIGTERM a meta' ordine perde la riga di log): se cuscino_watch tiene il lock questo
+  # giro si SALTA (rc 75) invece di sovrapporsi. Niente kill/timeout sul comando.
+  flock -n -E 75 logs/.conto.lock uv run python scripts/live/book_execute.py --execute; rc=$?
+  [ $rc -eq 75 ] && echo "  book: SALTATO — conto occupato da un altro giro (logs/.conto.lock)"
   # Libro di bordo: materializza in data/live/trades.db (che il backup COPRE) l'ora vera dei
   # fill, che altrimenti vive solo in questo log — gitignored, fuori dal backup e ruotabile.
-  # Va DOPO book_execute: dentro lo stesso blocco vede le righe appena scritte.
+  # Va DOPO book_execute: dentro lo stesso blocco vede le righe appena scritte. Fuori dal lock:
+  # e' sola lettura, non tocca il conto.
   uv run python scripts/live/trades_db.py --sync --quiet || true
   echo "===== done $(date -u '+%H:%M:%SZ') ====="
 } >> logs/cron_book.log 2>&1

scripts/cron_cuscino.sh

@@ -5,11 +5,19 @@
 # PUO' INVIARE ORDINI: sotto zero di slack lancia usde_convert --esegui (vende USDE), con le
 # guardie dichiarate nel docstring di scripts/live/cuscino_watch.py; lo interruttore e' lo
 # stesso del libro (execution_enabled in config/live.json).
+# LOCK SUL CONTO (#36): cuscino_watch e book_execute (cron_book.sh, :47) sono separati solo dal
+# tempo e possono mandare ordini sullo stesso conto — giri fino a 659s sono stati misurati.
+# `flock -n` (mai `-w`: un'attesa farebbe scivolare il giro orario) su logs/.conto.lock: se
+# book_execute lo tiene, questo giro si SALTA (rc 75) invece di sovrapporsi.
 export PATH="/home/adriano/.local/bin:$PATH"
 cd /opt/docker/PythagorasGoal || exit 1
 mkdir -p logs
 {
   echo "===== $(date -u '+%Y-%m-%dT%H:%M:%SZ') cron_cuscino ====="
-  uv run python scripts/live/cuscino_watch.py --quiet || true
+  flock -n -E 75 logs/.conto.lock uv run python scripts/live/cuscino_watch.py --quiet; rc=$?
+  [ $rc -eq 75 ] && echo "  cuscino: SALTATO — conto occupato (logs/.conto.lock)"
+  # (#30) giro orario dei sorveglianti a max_age<=6h: notifica solo il NUOVO, marcatore in
+  # data/live/monitor_health_detto.json. Fuori dal lock: e' sola lettura, non tocca il conto.
+  uv run python scripts/live/monitor_health.py --quiet --solo-orari || true
   echo "===== done $(date -u '+%H:%M:%SZ') ====="
 } >> logs/cron_cuscino.log 2>&1

scripts/live/cuscino_watch.py

@@ -80,7 +80,9 @@ CONVERT = ROOT / "scripts" / "live" / "usde_convert.py"
 # dell'ordine imposto (revisione fable 10/09: a margine == preavviso ogni 🚨 era seguito da un ⚠️).
 RIPRISTINO_MIN_ORE = 6.0     # un tentativo di conversione (in qualunque verso) ogni 6 ore
 RITENTATIVO_FALLITO_ORE = 24.0   # da ECCEDENTE, dopo un fallimento: un tentativo al giorno
-TIMEOUT_CONVERT_S = 900      # usde_convert spazia gli ordini di 12s: 60 ordini = 12 min
+TIMEOUT_CONVERT_S = 900      # usde_convert: 12s di pausa fra ordini + le GET per giro (~14s/giro
+                              # coi 12s di pausa); 60 ordini ~= 840s: il tetto e' stretto sul caso
+                              # massimo, non largo come "60 ordini = 12 min" lasciava credere (#35 B10)
 
 
 def leggi(path: Path = STATE) -> list[dict]:
@@ -158,17 +160,56 @@ def puo_riconvertire(prev_records: list[dict], now_ms: int, execution_enabled: b
     return True, ""
 
 
+def _leggi_quota_usde() -> float | None:
+    """Quota USDE sul totale, STESSA formula di usde_watch.py:275 (usd/(usd+eq_usdc)), stessi
+    mattoni di main() qui sotto (_safe_client, account_summary, U.prezzo, U.valuta): nessuna
+    lettura nuova. None se il conto non e' leggibile — chi chiama tratta None come "non lo so",
+    non come zero (#35 B10)."""
+    client = _safe_client()
+    if client is None:
+        return None
+    try:
+        eq_usdc = float(client.account_summary("USDC")["equity"])
+        eq_usde = float(client.account_summary("USDE").get("equity") or 0)
+    except Exception:
+        return None
+    px, _ = U.prezzo(client)
+    usd, _ = U.valuta(eq_usde, px)
+    try:
+        return round(usd / (usd + eq_usdc), 4)
+    except ZeroDivisionError:
+        return None
+
+
 def riconverti(quota: float, timeout_s: float = TIMEOUT_CONVERT_S) -> dict:
-    """Lancia lo script VERO con le sue guardie. Ritorna esito + coda dell'output (P3)."""
+    """Lancia lo script VERO con le sue guardie. Ritorna esito + coda dell'output (P3).
+
+    Rilegge la quota USDE prima e dopo il subprocess, ANCHE su timeout o eccezione: un rc!=0 con
+    la quota mossa e' una conversione PARZIALE (fill gia' avvenuti prima del kill a timeout_s),
+    non "FALLITA" come diceva `ok = rc==0` (#35 B10). `esito` in "ok" | "parziale" | "fallita"* —
+    "ok" solo per compatibilita' resta anche in `ok` (bool)."""
+    quota_prima = _leggi_quota_usde()
     cmd = [sys.executable, str(CONVERT), "--quota", f"{quota:.4f}", "--esegui"]
     try:
         r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout_s, cwd=str(ROOT))
+        rc = r.returncode
         coda = (r.stdout + r.stderr).strip().splitlines()[-12:]
-        return dict(rc=r.returncode, ok=r.returncode == 0, coda=coda, cmd=" ".join(cmd[1:]))
     except subprocess.TimeoutExpired:
-        return dict(rc=None, ok=False, coda=[f"timeout dopo {timeout_s:.0f}s"], cmd=" ".join(cmd[1:]))
+        rc, coda = None, [f"timeout dopo {timeout_s:.0f}s"]
     except Exception as e:                        # noqa: BLE001 — si registra, non si ingoia
-        return dict(rc=None, ok=False, coda=[f"{type(e).__name__}: {e}"], cmd=" ".join(cmd[1:]))
+        rc, coda = None, [f"{type(e).__name__}: {e}"]
+    quota_dopo = _leggi_quota_usde()
+
+    if rc == 0:
+        esito = "ok"
+    elif quota_dopo is None:
+        esito = "fallita (quota dopo non leggibile)"
+    else:
+        mossa = quota_prima is not None and abs(quota_dopo - quota_prima) > 0.005
+        esito = "parziale" if mossa else "fallita"
+
+    return dict(rc=rc, ok=esito == "ok", esito=esito, quota_prima=quota_prima, quota_dopo=quota_dopo,
+               coda=coda, cmd=" ".join(cmd[1:]))
 
 
 def _safe_client():
@@ -239,17 +280,23 @@ def main() -> int:
             rec["riconversione"] = dict(tentata=False, verso=AZIONE[stato], quota=quota,
                                         motivo=f"--secco: avrei lanciato usde_convert --quota {quota:.4f} --esegui ({AZIONE[stato]})")
         else:
-            esito = riconverti(quota)
-            rec["riconversione"] = dict(tentata=True, verso=AZIONE[stato], quota=quota, **esito)
+            ris = riconverti(quota)
+            rec["riconversione"] = dict(tentata=True, verso=AZIONE[stato], quota=quota, **ris)
 
     # --- allarmi: alla transizione, marcatore scritto con l'esito dell'invio ---
     # ⚠️ Telegram e' in parse_mode HTML: un '<' nudo nel testo fa fallire l'invio (successo al
     # primo giro vero, 13:53Z del 10/09: «(< 10% del cuscino)»). Qui non si scrive mai '<'.
     def _esito_azione(r: dict) -> str:
-        if r.get("tentata"):
-            return (f"{r['verso']} " + ("ESEGUITA" if r.get("ok") else "FALLITA")
-                    + f" (quota bersaglio {r['quota']:.0%})")
+        if not r.get("tentata"):
             return f"{r.get('verso', 'conversione')} NON tentata: {r.get('motivo')}"
+        esito = r.get("esito", "ok" if r.get("ok") else "fallita")
+        if esito == "parziale":
+            qp, qd = r.get("quota_prima"), r.get("quota_dopo")
+            qp_s = f"{qp:.0%}" if qp is not None else "n/d"
+            qd_s = f"{qd:.0%}" if qd is not None else "n/d"
+            return f"{r['verso']} PARZIALE: quota da {qp_s} a {qd_s}"
+        return (f"{r['verso']} " + ("ESEGUITA" if esito == "ok" else "FALLITA")
+                + f" (quota bersaglio {r['quota']:.0%})")
 
     if transizione(prev, stato):
         if secco:
@@ -275,7 +322,9 @@ def main() -> int:
     elif rec["riconversione"] and rec["riconversione"].get("tentata") and not secco:
         # non e' una transizione ma e' un ordine mandato: si dice sempre
         r = rec["riconversione"]
-        rec["allertato"] = notify(f"📌 cuscino USDC: {r['verso']} " + ("eseguita" if r["ok"] else "FALLITA"),
+        esito = r.get("esito", "ok" if r.get("ok") else "fallita")
+        verbo = {"ok": "eseguita", "parziale": "PARZIALE"}.get(esito, "FALLITA")
+        rec["allertato"] = notify(f"📌 cuscino USDC: {r['verso']} {verbo}",
                                   {"quota bersaglio": f"{r['quota']:.0%}", "esito": " | ".join(r["coda"][-3:])})
     if rec["allertato"] is False and not secco:
         from src.live.notifier import ultimo_errore

scripts/live/edge_watch.py

@@ -140,7 +140,7 @@ def main() -> int:
         print(f"\n     {n_ok}/{len(r['protection'])} anni di sinistro superati · "
               f"fallimenti consecutivi in corso: {r['consec_fail']}")
     if r["alerts"]:
-        notify("🚨 EDGE WATCH — criterio di kill scattato sul book LIVE",
+        notify("🚨 EDGE WATCH — criterio di kill scattato sul book (serie backtest dei sleeve, non il P&L eseguito)",
                {f"alert {i+1}": a for i, a in enumerate(r["alerts"])})
         for a in r["alerts"]:
             print(f"  ALERT: {a}")

scripts/live/fee_watch.py

@@ -79,6 +79,7 @@ SOGLIA_OK = 5.0               # <= -> nessuna azione
 SOGLIA_AZIONE = 10.0          # >  -> rivedere il peso di SKH01
 D_SHARPE_PER_BPS = -0.017     # sensibilita' marginale del BOOK, misurata
 D_CAGR_PER_BPS = -0.0023
+SOGLIA_DIVERGENZA_BPS = 1.0   # #37 B9: |fee realizzata - tier pubblico| oltre questo -> notify
 
 
 def fetch_fees(instrument: str) -> dict:
@@ -209,14 +210,30 @@ def run() -> dict:
     changes = diff_vs(prev, cur)
     worst = max(c["taker_bps"] for c in cur.values())
     lvl, why = verdict(worst)
+    # issue #37 (B9). realized_fee_bps() girava SOLO dentro il blocco di stampa: in cron
+    # (--quiet, il caso normale) non girava mai e una fee realmente pagata diversa dal tier
+    # pubblico non avvisava nessuno. Qui si calcola SEMPRE ed entra nel record del giro
+    # (registra_giro, sotto). {} -> None: "non misurata" (nessun trade recente), mai "zero".
+    real = realized_fee_bps() or None
+    divergenze = []
+    if real:
+        for ins, bps in real.items():
+            tier = cur.get(ins, {}).get("taker_bps")
+            if tier is None:
+                continue
+            d = bps - tier
+            if abs(d) > SOGLIA_DIVERGENZA_BPS:
+                divergenze.append(
+                    f"{ins}: realizzata {bps:.2f} bps vs tier pubblico {tier:.2f} ({d:+.2f})")
     return dict(current=cur, changes=changes, worst_taker_bps=worst, level=lvl, reason=why,
-                first_read=not prev)
+                first_read=not prev, realized=real, divergenze=divergenze,
+                divergenza_gia_detta=bool(prev.get("divergenza_detta")))
 
 
 def main() -> int:
     r = run()
     quiet = "--quiet" in sys.argv
-    interessante = bool(r["changes"]) or r["level"] != "OK"
+    interessante = bool(r["changes"]) or r["level"] != "OK" or bool(r.get("divergenze"))
 
     if not quiet or interessante:
         print("=" * 86)
@@ -232,7 +249,7 @@ def main() -> int:
             print("  (prima lettura: stato inizializzato, nessun confronto possibile)")
         for c in r["changes"]:
             print(f"  CAMBIO: {c}")
-        real = realized_fee_bps()
+        real = r.get("realized")  # calcolato una volta in run(), non piu' qui (#37 B9)
         if real:
             print("\n  cross-check sui trade REALI del conto (fonte autorevole):")
             for ins, bps in real.items():
@@ -240,7 +257,8 @@ def main() -> int:
                 nota = ""
                 if tier is not None:
                     d = bps - tier
-                    nota = f"   (tier taker {tier:.2f} → {d:+.2f})" + ("  ⚠️ DIVERGE" if abs(d) > 1.0 else "")
+                    nota = (f"   (tier taker {tier:.2f} → {d:+.2f})"
+                            + ("  ⚠️ DIVERGE" if abs(d) > SOGLIA_DIVERGENZA_BPS else ""))
                 print(f"    {ins:<22}{bps:>9.2f} bps/lato effettivi{nota}")
         else:
             print("\n  cross-check sui trade reali: NON MISURATO (nessun trade recente leggibile)")
@@ -257,9 +275,31 @@ def main() -> int:
         })
         if not inviato and (not quiet or interessante):
             print("  ⚠️ invio FALLITO — stato NON salvato: il prossimo giro ritenta")
+    # issue #37 (B9): fee realmente pagata che diverge dal tier pubblico atteso -> notify UNA
+    # volta per episodio. Marcatore "divergenza_detta" nello stesso state.json (stessa regola di
+    # B8: si scrive solo se l'invio e' riuscito), azzerato quando la divergenza torna sotto
+    # soglia. E' un evento indipendente dal cambio di tier: se qui serve un invio e quello di B8
+    # e' appena fallito, save_state sotto non gira (stessa gate di B8) e il marcatore non si
+    # scrive — al giro dopo si ridice, ridondante ma mai persa, la stessa scelta di B8.
+    divergenze = r.get("divergenze") or []
+    gia_detta = bool(r.get("divergenza_gia_detta"))
+    marcatore = gia_detta
+    if divergenze and not gia_detta:
+        inviata_div = notify("💸 FEE WATCH — fee realizzata diverge dal tier pubblico", {
+            f"divergenza {i+1}": d for i, d in enumerate(divergenze)
+        })
+        if inviata_div:
+            marcatore = True
+        elif not quiet or interessante:
+            print("  ⚠️ invio divergenza FALLITO — marcatore non scritto: il prossimo giro ritenta")
+    elif not divergenze:
+        marcatore = False
+
+    stato_da_salvare = dict(r["current"])
+    stato_da_salvare["divergenza_detta"] = marcatore
     if not serve_invio or inviato:
-        save_state(r["current"])
-    registra_giro(RUNS, level=r["level"])
+        save_state(stato_da_salvare)
+    registra_giro(RUNS, level=r["level"], realized=r.get("realized"))
     return 2 if r["level"] == "AZIONE" else 0
 
 

scripts/live/monitor_health.py

@@ -10,20 +10,91 @@ monitor e la soglia di copertura 0.80 (la stessa del veto d'integrita' di DVOLSP
 
     uv run python scripts/live/monitor_health.py            # report
     uv run python scripts/live/monitor_health.py --quiet    # stampa/allerta solo se scatta
+    uv run python scripts/live/monitor_health.py --solo-orari   # (#30) solo i sorveglianti a
+        # max_age<=6h (cuscino_watch, venue_watch, collect_chain), notifica solo il NUOVO
+        # rispetto a data/live/monitor_health_detto.json — pensato per girare ogni ora
+        # (cron_cuscino.sh), non per sostituire il giro giornaliero completo.
 """
 from __future__ import annotations
 
+import json
 import sys
+from datetime import datetime, timezone
 from pathlib import Path
 
 ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT))
 
-from src.live.monitor_health import MIN_COVERAGE, alerts_from, check_all  # noqa: E402
+from src.live.monitor_health import (  # noqa: E402
+    MIN_COVERAGE,
+    alert_nuovi,
+    alerts_from,
+    check_all,
+    marcatore_aggiornato,
+    monitors_orari,
+)
 from src.live.notifier import notify                                      # noqa: E402
 
+# (#30) marcatore "gia' detto" del giro orario: {"detti": [<chiave alert>], "ts": ...}. Vive fuori
+# da src/ perche' e' I/O, non parte della misura pura.
+DETTO = ROOT / "data" / "live" / "monitor_health_detto.json"
+
+
+def _carica_detti(path: Path = DETTO) -> set[str]:
+    if not path.exists():
+        return set()
+    try:
+        return set(json.loads(path.read_text()).get("detti", []))
+    except Exception:
+        return set()
+
+
+def _salva_detti(detti: set[str], path: Path = DETTO) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(
+        {"detti": sorted(detti), "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")},
+        indent=2))
+
+
+def main_orari() -> int:
+    """(#30) Giro orario: solo i sorveglianti a max_age<=6h, notifica solo il NUOVO rispetto al
+    marcatore. Exit 2 se c'e' un alert (anche gia' detto), 0 altrimenti — il giro giornaliero
+    resta l'unico posto dove "nessun alert nuovo" vuol dire davvero "tutto OK"."""
+    quiet = "--quiet" in sys.argv
+    rows = check_all(ROOT, monitors=monitors_orari())
+    alerts = alerts_from(rows)
+    detti = _carica_detti(DETTO)   # DETTO letto qui, non nel default del parametro (monkeypatch)
+    nuovi = alert_nuovi(rows, detti)
+    aggiornato = marcatore_aggiornato(rows, detti)
+
+    if not quiet or alerts:
+        print("=" * 92)
+        print("  MONITOR HEALTH — giro orario (sorveglianti con max_age <= 6h)")
+        print("=" * 92)
+        for r in rows:
+            cov = f"{r['coverage']:.0%}" if r["coverage"] is not None else "-"
+            age = f"{r['age_h']:.0f}h" if r["age_h"] is not None else "-"
+            print(f"  {r['name']:<20}{r['status']:>10}{age:>13}{cov:>10}")
+
+    inviato = True
+    if nuovi:
+        msgs = alerts_from(nuovi)
+        inviato = notify("🩺 MONITOR HEALTH — un sorvegliante orario non sta registrando",
+                          {f"alert {i+1}": m for i, m in enumerate(msgs)})
+        for m in msgs:
+            print(f"  ALERT (nuovo): {m}")
+        if not inviato:
+            print("  ⚠️ invio FALLITO — marcatore NON aggiornato: il prossimo giro ritenta")
+    if not nuovi or inviato:
+        _salva_detti(aggiornato, DETTO)
+
+    return 2 if alerts else 0
+
 
 def main() -> int:
+    if "--solo-orari" in sys.argv:
+        return main_orari()
+
     rows = check_all(ROOT)
     alerts = alerts_from(rows)
 
@@ -55,15 +126,17 @@ def main() -> int:
     return 0
 
 
-USO = """uso: monitor_health.py [--quiet]
+USO = """uso: monitor_health.py [--quiet] [--solo-orari]
 
   monitor_health.py — i forward-monitor stanno registrando? Riporta e allerta, non tocca nulla.
   uv run python scripts/live/monitor_health.py   # report
   uv run python scripts/live/monitor_health.py --quiet   # stampa/allerta solo se scatta
+  uv run python scripts/live/monitor_health.py --solo-orari   # (#30) solo i sorveglianti a
+      max_age<=6h, notifica solo il NUOVO rispetto al marcatore
 
 Dettaglio nel docstring in testa al file."""
 
 if __name__ == "__main__":
     from src.live.cli import valida
-    valida("monitor_health.py", USO, flag=("--quiet",), con_valore=())
+    valida("monitor_health.py", USO, flag=("--quiet", "--solo-orari"), con_valore=())
     raise SystemExit(main())

scripts/live/telegram_daily.py

@@ -25,17 +25,63 @@ import pandas as pd
 ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT))
 
+from src.live.monitor_health import MONITORS
 from src.live.notifier import send
 
 HORIZONS = (30, 90, 180)
-GATES = [("STATARB-RESID", date(2026, 9, 27), "data/paper_statarb/returns.jsonl", "net_modeled"),
-         ("XSR01", date(2026, 10, 23), "data/paper_xsr/returns.jsonl", "net_modeled")]
+
+
+def _gate_date(nome_monitor: str, cablata: date) -> date:
+    """(#35 C14) la data del gate viene da monitor_health.MONITORS (`MonitorSpec.gate`), la
+    STESSA tabella che "METTE A RISCHIO il gate" negli alert dei sorveglianti: una fonte, non
+    una seconda dichiarazione duplicata di CLAUDE.md §4. Il campo e' prosa libera
+    ("YYYY-MM-DD — descrizione"): si legge la data solo se e' nella forma attesa. `cablata' e'
+    rete di sicurezza se il monitor sparisse o il campo cambiasse forma — non deve succedere per
+    i due gate qui sotto, oggi allineati (verificato a mano contro CLAUDE.md §4)."""
+    for m in MONITORS:
+        if m.name == nome_monitor and m.gate:
+            try:
+                return date.fromisoformat(m.gate[:10])
+            except ValueError:
+                break
+    return cablata
+
+
+GATES = [("STATARB-RESID", _gate_date("paper_statarb", date(2026, 9, 27)),
+          "data/paper_statarb/returns.jsonl", "net_modeled"),
+         ("XSR01", _gate_date("paper_xsr", date(2026, 10, 23)),
+          "data/paper_xsr/returns.jsonl", "net_modeled")]
 
 
 def _fmt_usd(x: float) -> str:
     return f"${x:,.2f}"
 
 
+def _soglia_accensione(segni: list[int], neg_refs: list[float]) -> float | None:
+    """(#35 C3) Quale riferimento, superato, accende TP01 (mean(segni) > 0)?
+
+    I segni sono sempre in {-1,+1}, mai 0 (trend_state li costruisce con `1 if px>refs[h]
+    else -1`: nessun pareggio possibile). Con k segni negativi da far diventare positivi, la
+    somma cresce di 2 per flip: serve il piu' piccolo k tale che sum(segni)+2k > 0. La soglia
+    e' il k-esimo riferimento negativo piu' basso (k=1 -> il minimo, come prima del fix; k=2 ->
+    il secondo, ecc.) — non sempre il minimo: con TUTTI i riferimenti negativi, superare solo il
+    piu' basso lascia la media <= 0 e TP01 a 0 (bug segnalato in #35 C3).
+
+    None se la media e' gia' positiva (nessuna soglia da dare) o se k supera il numero di
+    riferimenti negativi disponibili (non dovrebbe accadere: con n segni, flippare tutti gli n
+    negativi porta comunque la somma a +n > 0, quindi k <= len(neg_refs) sempre — resta una
+    guardia difensiva, non un caso noto)."""
+    s = sum(segni)
+    if s > 0 or not neg_refs:
+        return None
+    k = 0
+    while s + 2 * k <= 0:
+        k += 1
+    if k > len(neg_refs):
+        return None
+    return sorted(neg_refs)[k - 1]
+
+
 def trend_state() -> list[str]:
     """Stato TSMOM per asset + quanto manca all'accensione (2 orizzonti su 3 positivi).
 
@@ -59,9 +105,9 @@ def trend_state() -> list[str]:
             tgt = max(0.0, m)
             line = f"  {a} {px:,.0f} · segni " + "/".join(f"{sg[h]:+d}" for h in HORIZONS) + \
                    f" → TP01 {tgt:.2f}"
-            neg = [(h, refs[h]) for h in HORIZONS if sg[h] < 0]
-            if neg and tgt == 0.0:
-                h, r = min(neg, key=lambda t: t[1])
+            if tgt == 0.0:
+                r = _soglia_accensione(list(sg.values()), [refs[h] for h in HORIZONS if sg[h] < 0])
+                if r is not None:
                     line += f"\n     accensione a {r:,.0f} ({(r/px-1)*100:+.1f}%)"
             out.append(line)
         except Exception as e:
@@ -69,22 +115,41 @@ def trend_state() -> list[str]:
     return out
 
 
-def book_state() -> list[str]:
-    """Conto reale, posizioni e target netto correnti. Sola lettura."""
+def book_state() -> tuple[list[str], bool | None]:
+    """Conto reale, posizioni e target netto correnti. Sola lettura.
+
+    (#35 C14) Ritorna anche se il libro e' DAVVERO flat (nessuna posizione, nessun target: stessa
+    condizione che questa funzione gia' usa riga per riga per stampare "flat" invece di "pos ->
+    target"): serve a `build()` per decidere se l'header "Perche' non opera" e' vero o
+    fuorviante, SENZA una lettura nuova — None se lo stato del libro non e' leggibile (in quel
+    caso non sappiamo dire se opera)."""
     try:
         from src.live.book import book_report
         r = book_report()
         eq = r.get("equity")
         lines = [f"  equity {_fmt_usd(eq)}" if eq else "  equity non leggibile"]
+        flat = True
         for a in r.get("assets", []):
             pos, net = a.get("position_usd", 0.0), a.get("net_target", 0.0)
-            stato = "flat" if abs(pos) < 1 and abs(net) < 1 else f"pos {_fmt_usd(pos)} → target {_fmt_usd(net)}"
+            is_flat = abs(pos) < 1 and abs(net) < 1
+            flat = flat and is_flat
+            stato = "flat" if is_flat else f"pos {_fmt_usd(pos)} → target {_fmt_usd(net)}"
             lines.append(f"  {a['asset']}: {stato} (TP {a.get('tp_frac', 0):+.2f} · SKH {a.get('skh_sign', 0):+d})")
         if r.get("skh_error"):
             lines.append(f"  ⚠️ SKH feed KO: {r['skh_error']}")
-        return lines
+        return lines, flat
     except Exception as e:
-        return [f"  ⚠️ stato libro non leggibile: {type(e).__name__}: {e}"]
+        return [f"  ⚠️ stato libro non leggibile: {type(e).__name__}: {e}"], None
+
+
+def _fmt_giorni(giorni: int) -> str:
+    """(#35 C14) tre casi distinti, non un segno "−" cablato davanti a un intero che puo' essere
+    negativo: con giorni=-1 quel cablaggio stampava "(−-1g)", un doppio segno illeggibile."""
+    if giorni > 0:
+        return f"-{giorni}g"
+    if giorni == 0:
+        return "oggi"
+    return f"scaduto da {-giorni}g"
 
 
 def monitors() -> list[str]:
@@ -92,20 +157,21 @@ def monitors() -> list[str]:
     for name, dec, path, key in GATES:
         p = ROOT / path
         giorni = (dec - date.today()).days
+        gate_txt = f"gate {dec} ({_fmt_giorni(giorni)})"
         if not p.exists():
-            out.append(f"  {name}: in attesa della prima barra · gate {dec} (−{giorni}g)")
+            out.append(f"  {name}: in attesa della prima barra · {gate_txt}")
             continue
         try:
             rows = [json.loads(x) for x in p.read_text().splitlines() if x.strip()]
             if not rows:
-                out.append(f"  {name}: 0 barre · gate {dec} (−{giorni}g)")
+                out.append(f"  {name}: 0 barre · {gate_txt}")
                 continue
             r = np.array([x[key] for x in rows], float)
             sh = float(r.mean() / r.std() * np.sqrt(365)) if r.std() > 0 else 0.0
             tot = float(np.prod(1 + r) - 1) * 100
-            out.append(f"  {name}: {len(rows)}g · {tot:+.2f}% · Sh {sh:+.2f} · gate {dec} (−{giorni}g)")
+            out.append(f"  {name}: {len(rows)}g · {tot:+.2f}% · Sh {sh:+.2f} · {gate_txt}")
         except Exception:
-            out.append(f"  {name}: log illeggibile · gate {dec} (−{giorni}g)")
+            out.append(f"  {name}: log illeggibile · {gate_txt}")
     return out
 
 
@@ -129,8 +195,13 @@ def last_trade() -> str:
 def build() -> str:
     oggi = datetime.now(timezone.utc).strftime("%d/%m %H:%M UTC")
     L = [f"📊 <b>Book Deribit — {oggi}</b>", "", "<b>Conto</b>"]
-    L += book_state()
-    L += ["", "<b>Perche' non opera</b>"]
+    righe_conto, flat = book_state()
+    L += righe_conto
+    # (#35 C14) "Perche' non opera" e' vero solo se il libro e' DAVVERO flat (nessuna posizione,
+    # nessun target: gia' letto sopra da book_state, nessuna lettura nuova). Se non e' flat (ha
+    # posizioni o un target diverso da zero) o non si sa (book non leggibile), l'header fisso
+    # mentirebbe: si usa un titolo neutro.
+    L += ["", "<b>Perche' non opera</b>" if flat else "<b>Stato trend</b>"]
     L += trend_state()
     L += ["", "<b>Ordini</b>", last_trade()]
     L += ["", "<b>Monitor in osservazione</b>"]

scripts/live/trades_db.py

@@ -17,6 +17,7 @@ ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT))
 
 from src.live import tradesdb as T  # noqa: E402
+from src.live.notifier import notify  # noqa: E402
 
 
 def sync(verbose: bool = True) -> dict:
@@ -31,10 +32,35 @@ def sync(verbose: bool = True) -> dict:
     T.set_meta(con, "ultimo_sync", T.ora())
     tot = con.execute("SELECT COUNT(*) c FROM fills").fetchone()["c"]
     tot_e = con.execute("SELECT COUNT(*) c FROM equity").fetchone()["c"]
+
+    # issue #37 (C11). Dal 2026-08-24T01:07Z (T.CUTOFF_TS_VERO) book_executions.jsonl porta l'ora
+    # VERA per-fill (docstring di tradesdb.py), ma `sync()` popola il DB SOLO da cron_book.log —
+    # l'arricchimento dal jsonl e' TradingBook#11, non qui. Un conteggio diverso in quella
+    # finestra segnala che una delle due fonti sta perdendo fill. Notify UNA volta per episodio:
+    # marcatore in `meta` (stessa regola di fee_watch #37 B9 — scritto solo se l'invio riesce,
+    # azzerato quando i conteggi tornano uguali).
+    righe_jsonl = T.parse_executions_jsonl(T.EXEC_JSONL.read_text()) if T.EXEC_JSONL.exists() else []
+    n_jsonl_dal = sum(1 for r in righe_jsonl if str(r.get("ts_utc", "")) >= T.CUTOFF_TS_VERO)
+    n_db_dal = con.execute("SELECT COUNT(*) c FROM fills WHERE ts_utc >= ?",
+                           (T.CUTOFF_TS_VERO,)).fetchone()["c"]
+    diverge = n_jsonl_dal != n_db_dal
+    gia_detto = T.get_meta(con, "conteggio_diverso_detto") == "1"
+    if diverge:
+        msg = f"⚠️ trades.db: {n_jsonl_dal} fill nel jsonl, {n_db_dal} nel DB dal 24/08"
+        if verbose:
+            print(f"  {msg}")
+        if not gia_detto:
+            if notify("⚠️ trades.db — conteggio fill diverso dal jsonl dal 24/08",
+                      {"fill nel jsonl": n_jsonl_dal, "fill nel DB": n_db_dal}):
+                T.set_meta(con, "conteggio_diverso_detto", "1")
+    elif gia_detto:
+        T.set_meta(con, "conteggio_diverso_detto", "0")
+
     if verbose:
         print(f"  fills: +{n_f} (totale {tot}) | equity: +{n_e} (totale {tot_e}) | round-trip {n_r}")
     con.close()
-    return dict(nuovi_fill=n_f, nuovi_equity=n_e, roundtrip=n_r, totale_fill=tot)
+    return dict(nuovi_fill=n_f, nuovi_equity=n_e, roundtrip=n_r, totale_fill=tot,
+                jsonl_dal_24_08=n_jsonl_dal, db_dal_24_08=n_db_dal)
 
 
 def reconcile() -> None:
@@ -42,8 +68,8 @@ def reconcile() -> None:
     fills = [f for r in runs for f in r.fills]
     righe = T.parse_executions_jsonl(T.EXEC_JSONL.read_text()) if T.EXEC_JSONL.exists() else []
     rec = T.reconcile(fills, righe)
-    print(f"\n  cron_book.log      : {len(fills)} fill (con ORA VERA)")
-    print(f"  book_executions    : {len(righe)} righe (senza ora)")
+    print(f"\n  cron_book.log      : {len(fills)} fill (ora di inizio giro)")
+    print(f"  book_executions    : {len(righe)} righe (ora VERA per-fill dal 24/08, #37 C11)")
     print(f"  concordano         : {len(rec['ok'])}")
     print(f"  solo nel log       : {len(rec['solo_log'])}")
     print(f"  solo nel jsonl     : {len(rec['solo_jsonl'])}")

scripts/live/usde_convert.py

@@ -61,6 +61,18 @@ reazione sbagliata al sintomo giusto: consuma i tentativi senza toccare la causa
 
     uv run python scripts/live/usde_convert.py --quota 0.64            # piano, nessun ordine (0,64 = bersaglio di cuscino_watch)
     uv run python scripts/live/usde_convert.py --quota 0.64 --esegui   # manda l'ordine
+
+LOCK SUL CONTO (#36). Un lancio a MANO con --esegui condivide il conto con cuscino_watch e
+book_execute (cron_cuscino.sh/cron_book.sh, entrambi dentro `flock -n logs/.conto.lock`): senza
+lo stesso lock un ordine a mano nello stesso minuto del cron puo' convertire due volte verso
+bersagli diversi (limite D5 gia' dichiarato sopra). Si lancia quindi con:
+
+    flock logs/.conto.lock uv run python scripts/live/usde_convert.py --quota 0.64 --esegui
+
+⚠️ Qui NON senza `-n`: un lancio a mano puo' aspettare il turno invece di saltare. Questo script
+stesso NON prende il lock (vedi cuscino_watch.py): se lo facesse e girasse come figlio di
+cuscino_watch (che il lock lo tiene gia'), un secondo `flock -n` lo troverebbe occupato e
+fallirebbe subito, per costruzione, ogni singola volta.
 """
 from __future__ import annotations
 

src/live/analista.py

@@ -206,9 +206,22 @@ def testata(voce: dict) -> str:
     p, lb = voce.get("pnl") or {}, voce.get("libro") or {}
     dq, cum = p.get("delta_equity"), p.get("cumulato_da_arming")
     eq = p.get("equity_fine")
+    # (#35 C8) il cron gira alle 00:30 sul giorno CHIUSO: "oggi" e' ieri -> "nel giorno". E se
+    # ci sono movimenti di capitale CERTI (mov_giorno; bool() come journal.scorporo: assente o
+    # 0 -> nessun movimento) la cifra da mostrare e' trading_giorno (al netto), non l'equity
+    # grezza: altrimenti un versamento si legge come P&L (successo il 25/08, +$1.399 letti come
+    # profitto). Se trading_giorno non e' calcolabile si resta su delta_equity, ma con l'etichetta
+    # nuova: non e' "oggi", e non e' comunque il trading al netto.
+    dq_t = p.get("trading_giorno")
+    if p.get("mov_giorno") and dq_t is not None:
+        nota_dq = f"  ({dq_t:+.2f} nel giorno, di cui {p['mov_giorno']:+.2f} movimenti)"
+    elif dq is not None:
+        nota_dq = f"  ({dq:+.2f} nel giorno)"
+    else:
+        nota_dq = ""
     r = [f"<b>📓 Giornale — {_b(voce['giorno'])}</b>"]
     r.append("  equity " + (f"<b>${eq:,.2f}</b>" if eq is not None else "n/d")
-             + ("" if dq is None else f"  ({dq:+.2f} oggi)")
+             + nota_dq
              + ("" if cum is None else f"  ·  dall'arming {cum:+.2f}"
                 + (f" (di cui {p['mov_arming']:+.2f} versati -> trading "
                    f"{p['trading_da_arming']:+.2f})"
@@ -228,15 +241,22 @@ def per_telegram(voce: dict, analisi: str, stato: str, motivi: list[str] | None
     """
     testo = testata(voce)
     if analisi.strip():
-        corpo = _b(analisi.strip())
-        corpo = re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", corpo)      # il grassetto del giornale
+        grezzo = analisi.strip()
         if stato == "sospetta":
             testo += "\n\n⚠️ <i>controllo sui numeri: sospetta — " + _b("; ".join(motivi or [])) + "</i>"
-        testo += "\n\n" + corpo
+        # (#35 C13) il taglio va sul GREZZO, PRIMA di _b (html.escape) e PRIMA di **-><b>: un
+        # taglio sul testo gia' composto puo' spezzare un'entita' (es. "&am") o lasciare un <b>
+        # aperto, e Telegram rifiuta l'INTERO messaggio ("can't parse entities") — il taglio si
+        # perde tutto, non si accorcia. Il marcatore si aggiunge dopo, fuori dall'escape (e'
+        # testo nostro, senza `&`/`<`: non ha bisogno di escape).
+        taglio = "\n\n<i>[…tagliato: la voce intera e' in docs/journal/%s.md]</i>" % voce["giorno"]
+        budget = LIMITE_TELEGRAM - len(testo) - len("\n\n") - len(taglio)
+        tagliata = len(grezzo) > budget
+        if tagliata:
+            grezzo = grezzo[:max(budget, 0)]
+        corpo = re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", _b(grezzo))  # il grassetto del giornale
+        testo += "\n\n" + corpo + (taglio if tagliata else "")
     else:
         testo += ("\n\n⚠️ <i>analisi non disponibile (" + _b(stato) + "): "
                   + _b("; ".join(motivi or ["motivo non registrato"])) + "</i>")
-    if len(testo) > LIMITE_TELEGRAM:
-        taglio = "\n\n<i>[…tagliato: la voce intera e' in docs/journal/%s.md]</i>" % voce["giorno"]
-        testo = testo[:LIMITE_TELEGRAM - len(taglio)] + taglio
     return testo

src/live/book.py

@@ -4,8 +4,8 @@ TP01 e SKH01 tradano lo STESSO strumento (BTC/ETH _USDC-PERPETUAL). Su Deribit e
 posizione netta per strumento per conto -> non si possono tenere due gambe separate: si combinano in
 software in un unico target netto, e si manda UN ordine per asset per raggiungerlo.
 
-Formula (preserva il budget di rischio $300/asset, split 75/25; coerente col blend di ritorni
-deribit_book_sleeves = 0.75*TP01 + 0.25*SKH01):
+Formula (preserva il budget di rischio per-asset di config (CAP sotto: `max_notional_per_asset_usd`
+o `_frac`), split 75/25; coerente col blend di ritorni deribit_book_sleeves = 0.75*TP01 + 0.25*SKH01):
 
     net_target_usd[asset] = clamp( WEIGHT * E * (W_TP01 * tp_frac + W_SKH * skh_sign), ±CAP )
 
@@ -13,8 +13,8 @@ deribit_book_sleeves = 0.75*TP01 + 0.25*SKH01):
     tp_frac        = target TP01 (causale, >=0, long-flat)              da TrendPortfolio.current_target
     skh_sign       = +1 long / -1 short / 0 flat                        da _skyhook_positions (book 230m)
     E              = equity reale del conto (fallback paper se offline)
-    CAP            = cap notional per-asset. Fisso = max_notional_per_asset_usd ($300), OPPURE
-                     DINAMICO = equity * max_notional_per_asset_frac (es. 0.5 -> equity/2) se la
+    CAP            = cap notional per-asset. Fisso = `max_notional_per_asset_usd` di config, OPPURE
+                     DINAMICO = equity * `max_notional_per_asset_frac` (es. 0.5 -> equity/2) se la
                      frazione e' presente in config -> il cap cresce col capitale e non strozza il
                      book quando depositi (frontiera 2026-07-03). Dinamico SOLO con equity reale
                      fidata; su fallback/offline si torna al cap fisso (protezione downside).
@@ -136,6 +136,42 @@ def _scala(real_equity: float | None = None, eq_fallback: str | None = None,
     return s
 
 
+def leva_tetto(cfg: dict | None = None) -> tuple[float, str]:
+    """Leva lorda massima che il libro PUO' prendere. UNA formula, non tre (#28): prima era
+    ridichiarata in `usde.frazione_cuscino` (N_ASSET fisso, senza scala) e nella regola `leva` del
+    giornale (senza scala) -> a scala 1,25 il cuscino USDC risultava 0,30 dell'equity invece di
+    0,375, e la regola `leva` mostrava un tetto diverso da quello che il libro rispetta davvero.
+
+    leva = len(ASSETS) x min(WEIGHT, frac) x scala, al piu' LEVA_LORDA_MAX. -> (leva, come).
+    - frac = cfg["max_notional_per_asset_frac"] se presente, altrimenti WEIGHT. Il min(WEIGHT, frac)
+      viene dal fatto che `book_net_target` non supera mai weight*equity*(W_TP01+W_SKH) =
+      weight*equity (i due pesi sommano a 1): |raw| <= weight*equity, quindi |net_target| <=
+      scala*min(cap, weight*equity) -- un frac oltre WEIGHT alza il cap nominale ma non la leva
+      REALE che il libro puo' esprimere.
+    - scala = la scala DICHIARATA (`_scala(real_equity=1.0, cfg=cfg)`, gia' validata): un tetto
+      descrive quanto la config AUTORIZZA, non quanto un giro con equity illeggibile ha preso.
+    - Su `ScalaNonAutorizzata` (config fuori scaletta o fuori tetto) NON si solleva: un
+      sorvegliante (`cuscino_watch`) non deve morire proprio quando la config e' sbagliata
+      (verifica fable B5) -> si ritorna il tetto di CODICE, dichiarando l'anomalia in `come`.
+    """
+    if cfg is None:
+        cfg = json.loads(CONFIG.read_text()) if CONFIG.exists() else {}
+    try:
+        scala = _scala(real_equity=1.0, cfg=cfg)
+    except ScalaNonAutorizzata as e:
+        breve = str(e).splitlines()[0]
+        breve = breve if len(breve) <= 80 else breve[:77] + "..."
+        return LEVA_LORDA_MAX, (f"config fuori scaletta/tetto ({breve}): "
+                                 f"tetto di codice LEVA_LORDA_MAX")
+    raw_frac = cfg.get("max_notional_per_asset_frac")
+    frac = float(raw_frac) if raw_frac is not None else WEIGHT
+    frac_eff = min(WEIGHT, frac)
+    leva = len(ASSETS) * frac_eff * scala
+    base = f"{len(ASSETS)} asset x frac {frac_eff} x scala {scala} = {leva:.2f}x"
+    if leva > LEVA_LORDA_MAX + 1e-12:
+        return LEVA_LORDA_MAX, f"{base}, troncato al tetto di codice {LEVA_LORDA_MAX}x"
+    return leva, base
+
 
 def _read_watermark() -> float | None:
     """Ultima equity REALE osservata sul conto. None se mai vista o file illeggibile."""
@@ -269,6 +305,7 @@ def book_report(offline: bool = False, equity_override: float | None = None,
     live_feed: se True usa il feed 5m fresco effimero per SKH01 (src/live/livefeed.fresh_5m) ->
     segnale 230m all'ultima barra chiusa, per l'esecuzione reale. Default False (feed certificato,
     deterministico: dashboard/test). TP01 resta sul certificato (è giornaliero)."""
+    cfg = json.loads(CONFIG.read_text()) if CONFIG.exists() else {}
     sh = shadow_report(offline=offline, equity_override=equity_override)
     equity = sh["equity"]
     # registra l'equity reale quando e' leggibile: e' cio' che rende sicuro il cap di
@@ -315,7 +352,7 @@ def book_report(offline: bool = False, equity_override: float | None = None,
         net = book_net_target(tp_frac, sign, equity, cap, scala=scala)
         cur = float(a_rec["position_usd"])
         mark = a_rec["mark"]
-        order = build_book_order(inst, net, cur, mark, min_usd=5.0)
+        order = build_book_order(inst, net, cur, mark, min_usd=float(cfg.get("min_order_usd", 5.0)))
         if order:
             orders.append(order)
         assets.append(dict(

src/live/deribit.py

@@ -1,13 +1,16 @@
-"""Accesso Deribit MAINNET in SOLA LETTURA (via Cerbero MCP) + costruttore ordini deterministico.
+"""Accesso Deribit MAINNET (via Cerbero MCP) + costruttore ordini deterministico. La classe
+`DeribitRead` qui e' in SOLA LETTURA (nessun metodo di trading, by design); l'esecuzione REALE la
+estende in un modulo separato (`src.live.execution.DeribitTrader`) — non e' piu' un progetto
+futuro (#35 A9): e' ARMATA e LIVE su Deribit mainnet dal 2026-06-23 (`execution_enabled` in
+config/live.json), usata da `scripts/live/book_execute.py`.
 
-Serve lo SHADOW MODE di TP01 (`scripts/live/live_trend.py`): legge prezzi / conto / posizioni REALI
-dal mainnet (token `.env.mainnet`) e costruisce gli ordini di ribilancio **senza inviarli**. Qui NON
-esiste alcun metodo di trading — by design: l'unica via per piazzare ordini sara' un modulo separato,
-abilitato esplicitamente, dopo la validazione shadow + micro-test.
+Da qui vengono prezzi / conto / posizioni REALI (token `.env.mainnet`): li usa sia il libro live
+(TP01+SKH01, via src.live.book + execution) sia lo shadow mode e i monitor di sorveglianza.
 
-Disciplina del progetto: **testnet FUORI** (feed farlocco, causa del reset v2.0.0). Solo mainnet reale,
-e in questa fase solo in lettura. I contratti sono ristretti a BTC/ETH-PERPETUAL (inverse: `amount`
-in USD notional, step verificato su Deribit: BTC $10, ETH $1).
+Disciplina del progetto: **testnet FUORI** (feed farlocco, causa del reset v2.0.0). Solo mainnet
+reale. I contratti ESEGUITI sono i perp LINEARI USDC (`_USDC-PERPETUAL`, non inverse): `amount` in
+base-coin (BTC/ETH), margine e P&L in USDC — vedi `_CONTRACT` sotto per step/tick/minimi per
+strumento.
 """
 from __future__ import annotations
 

src/live/execution.py

@@ -23,10 +23,11 @@ from dataclasses import dataclass
 from src.live.deribit import DISASTER_LABEL, DeribitRead, disaster_stop_price, notional_to_amount, quantize_price
 
 # Conto USDC -> perp LINEARE USDC (amount in base-coin). MAX_AMOUNT = TETTO HARD anti-fat-finger di
-# DEFAULT (~$630/$430 su un conto ~$600), usato SOLO quando nessuno ha chiamato set_open_cap() (il
-# micro-test, che invia comunque size fissa minima). Sul libro il tetto per apertura lo deriva
-# set_open_cap() dal cap di config (config/live.json max_notional_per_asset_usd, #18): backstop
-# sopra il sizing di TP01/book, non il sizing operativo.
+# DEFAULT (in coin, sotto), usato SOLO quando nessuno ha chiamato set_open_cap() (il micro-test,
+# che invia comunque size fissa minima). Sul libro il tetto per apertura lo deriva set_open_cap()
+# dal cap di config (config/live.json max_notional_per_asset_usd, #18): backstop sopra il sizing
+# di TP01/book, non il sizing operativo. (#35 A9: via il "~$630/$430 su conto ~$600" cablato —
+# MAX_AMOUNT e' in coin, non in dollari, e non e' il tetto operativo da quando esiste set_open_cap.)
 ALLOWED = {"BTC_USDC-PERPETUAL", "ETH_USDC-PERPETUAL"}
 MAX_AMOUNT = {"BTC_USDC-PERPETUAL": 0.01, "ETH_USDC-PERPETUAL": 0.25}   # DEFAULT — vedi set_open_cap()
 FLAT_USD = 1.0   # |notional| < $1 = posizione considerata flat

src/live/journal.py

@@ -192,12 +192,10 @@ def movimenti_capitale(con, fino_ts: str, rows=None) -> dict:
         return dict(eventi=[], certi=0.0, ambigui=0.0, soglia=soglia, avvisi=avvisi)
     try:
         cfg = json.loads((PROJECT_ROOT / "config" / "live.json").read_text())
-        # la leva lorda massima che il libro puo' tenere: frac x n_asset x scala, mai oltre il
-        # tetto di CODICE (P1: derivata da `book`, non ridichiarata — la chiave di scala del
-        # 01/09 entra qui da sola)
-        scala = float(cfg.get("book_scale_k") or 1.0)
-        leva_tetto = min(float(cfg["max_notional_per_asset_frac"]) * len(ASSETS) * scala,
-                         float(_book.LEVA_LORDA_MAX))
+        # la leva lorda massima che il libro puo' tenere: UNA sola formula, in `book.leva_tetto`
+        # (P1: derivata da `book`, non ridichiarata — #28 ne aveva tre diverse fra qui,
+        # usde.frazione_cuscino e la regola `leva` sotto, e non tornavano)
+        leva_tetto = _book.leva_tetto(cfg)[0]
     except Exception:
         leva_tetto = float(_book.LEVA_LORDA_MAX)   # config illeggibile: il tetto di codice
     px = {a: _px_1h(a) for a in ASSETS} if candidati else {}
@@ -567,17 +565,17 @@ def lettura(con, voce: dict) -> list[dict]:
     # --- quanto capitale e' esposto ----------------------------------------------------------
     lv = lb.get("leva_lorda")
     if lv is not None:
-        # Il tetto si LEGGE da config/live.json. Ridichiararlo qui e' lo schema che questo
-        # progetto ha gia' pagato quattro volte (fee_watch, test di guardia sul cap, taratura
-        # venue_watch, indice del namespace): un sorvegliante che ridichiara il proprio
-        # bersaglio continua a passare il giorno che il bersaglio cambia.
+        # Il tetto si LEGGE da `book.leva_tetto`, non ridichiarato qui. Ridichiararlo e' lo schema
+        # che questo progetto ha gia' pagato quattro volte (fee_watch, test di guardia sul cap,
+        # taratura venue_watch, indice del namespace) — e una quinta, con una formula diversa da
+        # quella di `movimenti_capitale` e di `usde.frazione_cuscino`, e' quella corretta da #28.
+        from src.live import book as _book      # lazy: prende il tetto vivo, anche monkeypatchato
         try:
             cfg = json.loads((PROJECT_ROOT / "config" / "live.json").read_text())
-            cap_frac = float(cfg["max_notional_per_asset_frac"])
-            tetto = cap_frac * len(lb["asset"])
+            tetto, come = _book.leva_tetto(cfg)
             dire("leva", f"Leva lorda **{lv:.2f}x** su un tetto di {tetto:.2f}x "
-                         f"(`max_notional_per_asset_frac` {cap_frac:.2f} x {len(lb['asset'])} "
-                         f"asset, letto da `config/live.json`): margine {tetto - lv:.2f}x.")
+                         f"({come}, da `config/live.json` e `book.leva_tetto`): "
+                         f"margine {tetto - lv:.2f}x.")
         except Exception as e:
             dire("leva", f"Leva lorda **{lv:.2f}x**; tetto non verificabile "
                          f"({type(e).__name__} su config/live.json).", "attenzione")

src/live/monitor_health.py

@@ -49,6 +49,9 @@ import numpy as np
 MIN_COVERAGE = 0.80          # sotto: la serie ha buchi -> non ci si decide sopra un gate
 MIN_BARS_PER_JUDGE = 2       # con una barra sola non esiste ne' coda ne' copertura
 GRAZIA_PREMATURO_MIN = 5.0   # tolleranza fra chiusura della barra e mtime del file (clock/IO)
+SOGLIA_ORARI_H = 6.0         # (#30) max_age_h <= questa soglia = "sorvegliante orario": i tre
+                              # con max_age 3h (cuscino_watch, venue_watch, collect_chain) oggi,
+                              # derivato dalla tabella e non da una lista di nomi cablata
 
 
 @dataclass(frozen=True)
@@ -244,3 +247,46 @@ def alerts_from(rows: list[dict]) -> list[str]:
             msg += f" | METTE A RISCHIO il gate {r['gate']}"
         out.append(msg)
     return out
+
+
+# ===========================================================================
+# GIRO ORARIO (#30). Il giro giornaliero (cron_daily) resta com'e': report completo, avviso
+# sempre. I sorveglianti a max_age <= 6h (oggi cuscino_watch, venue_watch, collect_chain) si
+# scoprono fermi fino a 24h dopo se nessuno li guarda fuori da quel giro: qui si guardano ogni
+# ora, ma si notifica solo cio' che e' NUOVO rispetto all'ultimo avviso (niente Telegram a
+# raffica sullo stesso guasto fermo). Tutto PURO: il marcatore lo legge/scrive lo script.
+# ===========================================================================
+def monitors_orari(monitors: tuple[MonitorSpec, ...] = MONITORS,
+                    soglia_h: float = SOGLIA_ORARI_H) -> tuple[MonitorSpec, ...]:
+    """I monitor con cadenza di controllo ORARIA: max_age_h <= soglia_h. Derivato dalla tabella
+    `MONITORS`, non da una lista di nomi cablata: aggiungere un monitor orario lo include qui
+    senza toccare questa funzione."""
+    return tuple(m for m in monitors if m.max_age_h <= soglia_h)
+
+
+def chiave_alert(row: dict) -> str:
+    """Chiave stabile di un alert per il marcatore 'gia' detto': nome + stato, NON il testo
+    (`why` contiene l'eta' dell'ultima barra, che cambia a ogni giro anche se il guasto e' lo
+    stesso)."""
+    return f"{row['name']}:{row['status']}"
+
+
+def alert_nuovi(rows: list[dict], detti: set[str]) -> list[dict]:
+    """Le righe-alert (non OK, non NUOVO — vedi `alerts_from`) la cui chiave NON e' nel
+    marcatore `detti`: quelle davvero nuove, le sole da notificare."""
+    return [r for r in rows if r["status"] not in ("OK", "NUOVO") and chiave_alert(r) not in detti]
+
+
+def marcatore_aggiornato(rows: list[dict], detti: set[str]) -> set[str]:
+    """Il marcatore dopo questo giro: le chiavi dei monitor ancora in alert restano/si aggiungono,
+    quelle dei monitor tornati OK si TOLGONO — cosi' un nuovo guasto sullo stesso monitor riparte
+    con un nuovo avviso invece di restare 'gia' detto' per sempre. Non scrive nulla: lo fa lo
+    script, e solo se la notifica e' partita (stessa regola di fee_watch #29)."""
+    out = set(detti)
+    for r in rows:
+        chiavi_monitor = {k for k in out if k.startswith(f"{r['name']}:")}
+        if r["status"] == "OK":
+            out -= chiavi_monitor
+        elif r["status"] != "NUOVO":
+            out.add(chiave_alert(r))
+    return out

src/live/revisione.py

@@ -39,8 +39,16 @@ FINESTRA_GIORNALE_G = 7               # pagine del giornale lette
 FINESTRA_DIARI_G = 7                  # diari letti (la settimana: e' una revisione settimanale)
 FINESTRA_COMMIT_G = 14                # commit letti
 CARTELLA = Path("docs") / "revisioni"
-MAX_CHARS_VOCE = 90_000               # CLAUDE.md sta sotto; il resto e' molto piu' piccolo
+MAX_CHARS_VOCE = 90_000               # il resto delle voci e' molto piu' piccolo
+# #37 C12. CLAUDE.md e' arrivato a 89.590 caratteri (99,5% di MAX_CHARS_VOCE, 27/09): al primo
+# paragrafo aggiunto si sarebbe troncato. Tetto SUO, piu' largo — il vincolo vero resta
+# MAX_CHARS_TOTALE, applicato A PARTE in `materiale()` a prescindere dal tetto di ogni voce: la
+# somma dei tetti per-voce (che supera 400k) non e' il limite, lo e' il secondo giro di `_tronca`.
+MAX_CHARS_CLAUDE_MD = 150_000
 MAX_CHARS_TOTALE = 400_000
+SOGLIA_AVVISO_TETTO = 0.95            # avviso quando una voce supera questa frazione del SUO tetto
+_MARCA_VICINO_TETTO = "[… vicino al tetto:"   # marcatore condiviso: materiale() lo scrive, i
+                                                # consumatori (prompt, rapporto) lo rileggono
 MAX_NUMERI_LIBERI = 12                # una revisione confronta e sottrae: qualche numero suo e' lecito
 LIMITE_TELEGRAM = 3_500               # sotto i 4.096 di Telegram, con la testata
 TIMEOUT_S = 900                       # il materiale e' grande: 5 minuti non bastano sempre
@@ -89,6 +97,12 @@ def _tronca(testo: str, n: int | None = None) -> tuple[str, bool]:
     return testo[:n] + f"\n[… troncato a {n:,} caratteri su {len(testo):,}]", True
 
 
+def _cap_voce(nome: str) -> int:
+    """Tetto PER-VOCE: CLAUDE.md ne ha uno suo, piu' largo (#37 C12). Il totale
+    (MAX_CHARS_TOTALE) resta un vincolo A PARTE, applicato in `materiale()` a valle di questo."""
+    return MAX_CHARS_CLAUDE_MD if nome.startswith("CLAUDE.md") else MAX_CHARS_VOCE
+
+
 def _cmd(args: list[str], cwd: Path) -> str:
     """Uscita di un comando di SOLA LETTURA, o il motivo per cui non si legge (P5: «non vedo»
     non e' «va tutto bene»)."""
@@ -175,7 +189,18 @@ def materiale(root: Path, adesso: datetime) -> list[tuple[str, str, bool]]:
 
     out, tot = [], 0
     for nome, testo in voci:
-        t, tr = _tronca(testo)
+        cap = _cap_voce(nome)
+        # #37 C12: avviso PRIMA di troncare, quando la voce e' vicina al SUO tetto (non ancora al
+        # tetto totale, sotto): la nota entra nel testo, cosi' arriva sia al prompt del revisore
+        # sia — via _MARCA_VICINO_TETTO — al rapporto, come [TRONCATO] oggi. Il margine
+        # (`+ len(nota) <= cap`) e' voluto: la nota stessa non deve MAI far scattare il
+        # troncamento che sta segnalando in anticipo.
+        if len(testo) >= SOGLIA_AVVISO_TETTO * cap:
+            nota = (f"\n{_MARCA_VICINO_TETTO} {len(testo):,} di {cap:,} caratteri "
+                    f"({100 * len(testo) / cap:.0f}%)]")
+            if len(testo) + len(nota) <= cap:
+                testo = testo + nota
+        t, tr = _tronca(testo, cap)
         if tot + len(t) > MAX_CHARS_TOTALE:
             t, tr = _tronca(t, max(0, MAX_CHARS_TOTALE - tot)), True
         tot += len(t)
@@ -186,7 +211,8 @@ def materiale(root: Path, adesso: datetime) -> list[tuple[str, str, bool]]:
 def costruisci_prompt(voci: list[tuple[str, str, bool]], adesso: datetime) -> str:
     parti = [ISTRUZIONI, f"\nDATA DELLA REVISIONE: {adesso:%Y-%m-%d %H:%M}Z (usa questa per «oggi» e per le date dei gate).\n"]
     for nome, testo, tr in voci:
-        parti.append(f"\n=== {nome}{' [TRONCATO]' if tr else ''} ===\n{testo}\n")
+        tag = " [TRONCATO]" if tr else (" [VICINO AL TETTO]" if _MARCA_VICINO_TETTO in testo else "")
+        parti.append(f"\n=== {nome}{tag} ===\n{testo}\n")
     parti.append("\nScrivi ora la revisione, coi cinque titoli, e nient'altro.")
     return "\n".join(parti)
 
@@ -236,7 +262,8 @@ def rapporto(testo: str | None, errore: str | None, stato: str, motivi: list[str
             "- scritto da `scripts/live/revisione.py` (cron `scripts/cron_review.sh`); l'unico file "
             "scritto e' questo", "",
             "| voce | caratteri | troncata |", "|---|---|---|"]
-    coda += [f"| {n} | {len(t):,} | {'si' if tr else ''} |" for n, t, tr in voci]
+    coda += [f"| {n} | {len(t):,} | "
+             f"{'si' if tr else ('quasi' if _MARCA_VICINO_TETTO in t else '')} |" for n, t, tr in voci]
     return "\n".join(testa + corpo + coda) + "\n"
 
 
@@ -258,10 +285,14 @@ def per_telegram(testo: str | None, errore: str | None, stato: str, adesso: date
         return (f"<b>🚨 Revisione {CADENZA} NON eseguita</b> {adesso:%Y-%m-%d}\n"
                 f"{html.escape(errore)}\n<code>{html.escape(str(path))}</code>")
     sintesi = _sezione(testo or "", TITOLI[0]) or "(nessuna Sintesi nel rapporto)"
-    corpo = html.escape(sintesi, quote=False)
     tagliato = ""
-    if len(corpo) > LIMITE_TELEGRAM:
-        corpo, tagliato = corpo[:LIMITE_TELEGRAM], "\n[… sintesi tagliata: il resto e' nel file]"
+    # (#35 C13) il taglio va sul testo GREZZO, PRIMA di html.escape: tagliare dopo puo' spezzare
+    # un'entita' a meta' (es. "&am") e Telegram rifiuta l'intero messaggio ("can't parse
+    # entities") invece di accorciarlo. Il marcatore di taglio si aggiunge dopo, fuori dall'escape
+    # (e' testo nostro, senza `&`/`<`: non ha bisogno di escape).
+    if len(sintesi) > LIMITE_TELEGRAM:
+        sintesi, tagliato = sintesi[:LIMITE_TELEGRAM], "\n[… sintesi tagliata: il resto e' nel file]"
+    corpo = html.escape(sintesi, quote=False)
     return (f"<b>🔍 Revisione {CADENZA}</b> {adesso:%Y-%m-%d} · guardia numeri: {stato}\n\n"
             f"{corpo}{tagliato}\n\n<code>{html.escape(str(path))}</code>")
 
@@ -312,7 +343,8 @@ def scrivi(root: Path, adesso: datetime | None = None, modello: str = MODELLO, *
         print(f"  revisione {esito['data']}  modello {modello}  materiale {len(voci)} voci, "
               f"{len(prompt):,} caratteri" + ("  [SECCO: nessuna chiamata, nessuna scrittura]" if secco else ""))
         for n, t, tr in voci:
-            print(f"    {len(t):>8,}  {n}{'  [troncata]' if tr else ''}")
+            tag = "  [troncata]" if tr else ("  [vicino al tetto]" if _MARCA_VICINO_TETTO in t else "")
+            print(f"    {len(t):>8,}  {n}{tag}")
     if secco:
         return esito
 

src/live/tradesdb.py

@@ -2,20 +2,31 @@
 
 PERCHE' ESISTE
 --------------
-`data/live/book_executions.jsonl` registra i fill, ma il suo `ts_utc` e' la data della
-**barra di segnale** (`book_execute.py`: `pd.Timestamp(r['last_data'])`), non l'ora del fill:
-19 righe su 19 a `00:00:00`. L'ora vera sta solo in `logs/cron_book.log` — che e' **gitignored,
-non nel backup e ruotabile**. Quindi oggi la cronologia reale del libro live vive in un file
-che una rotazione cancella, e nessuno se ne accorgerebbe.
+`logs/cron_book.log` e' **gitignored, non nel backup e ruotabile**, ed e' la fonte che `sync()`
+usa per popolare il DB: una rotazione lo cancellerebbe senza che nessuno se ne accorga.
 
 Questo modulo MATERIALIZZA quella cronologia in `data/live/trades.db` (sqlite), che sta dentro
 il perimetro gia' coperto dal backup rotativo della VPS.
 
+⚠️ CORREZIONE (#37 C11, #35 A9). Fino al 2026-08-23 `data/live/book_executions.jsonl` scriveva in
+`ts_utc` la data della **barra di segnale** (`pd.Timestamp(r['last_data'])`), non l'ora del fill:
+19 righe su 19 a `00:00:00`. Dal 2026-08-24 01:07Z `book_execute.py` (`_registra_fill`) scrive
+l'ora VERA del fill (`datetime.now(timezone.utc)` al momento dell'ordine, non piu' la barra),
+UNA per fill — con il prezzo a piena precisione (non arrotondato come nella stampa del log) e il
+campo `pos_after` (oggi sempre `None` qui: la posizione post-rebalance si legge dopo, best-effort,
+e non viene riscritta nel jsonl — vedi il commento in `_registra_fill`). E' quindi, dal 24/08 in
+poi, la fonte con l'ora PIU' precisa per-fill — ma resta usata solo da `reconcile()`: arricchire
+`fills` con ts/prezzo/pos_after dal jsonl e' uno SVILUPPO non fatto qui (TradingBook#11).
+
 FONTI, in ordine di autorita' — e nessuna delle tre e' completa da sola
 ----------------------------------------------------------------------
-1. `logs/cron_book.log`  -> ORA VERA del giro + contesto (equity, tp_frac, skh, target, posizione)
-                            e la riga di fill `-> BUY 0.0008 @ $78,094.2 fee 0.02187 (OK)`.
-2. `data/live/book_executions.jsonl` -> gli stessi fill con piu' cifre, ma senza ora.
+1. `logs/cron_book.log`  -> l'ora di INIZIO GIRO (una per run, condivisa da tutti i fill di quel
+                            run, non per-fill) + contesto (equity, tp_frac, skh, target,
+                            posizione) e la riga di fill
+                            `-> BUY 0.0008 @ $78,094.2 fee 0.02187 (OK)`. FONTE usata da `sync()`.
+2. `data/live/book_executions.jsonl` -> gli stessi fill, con l'ora VERA per-fill dal 2026-08-24
+                            01:07Z (prima: la data della barra di segnale, 00:00:00), prezzo a
+                            piena precisione e `pos_after`. Entra SOLO in `reconcile()`.
 3. venue (`DeribitRead.trade_history`) -> autorevole su `order_id` e timestamp in ms,
                             **ma tronca**: al 2026-08-23 ritorna 1 trade su BTC e 0 su ETH.
 
@@ -42,6 +53,10 @@ EXEC_JSONL = PROJECT_ROOT / "data" / "live" / "book_executions.jsonl"
 
 # Data di armamento dell'esecuzione reale (CLAUDE.md, 2026-06-20).
 ARMING = "2026-06-20"
+# Da qui in poi book_executions.jsonl porta l'ora VERA del fill in `ts_utc` (fix #35 A9, docstring
+# sopra): soglia per il controllo di conteggio jsonl-vs-DB in scripts/live/trades_db.py:sync()
+# (#37 C11). Prima di questo istante il campo e' la barra di segnale, non comparabile.
+CUTOFF_TS_VERO = "2026-08-24T01:07:00+00:00"
 
 SCHEMA = """
 CREATE TABLE IF NOT EXISTS fills (

src/live/usde.py

@@ -104,12 +104,12 @@ def valuta(eq_usde: float, px: float | None) -> tuple[float, str]:
 # il saldo USDC va negativo alla prima perdita del libro e Deribit lo finanzia allo 0,05%/GIORNO
 # (KB "Cross collateral specifications": 18,25%/anno, 4,3x la resa USDE). Il cuscino richiesto e'
 # il disaster-SL sulla MASSIMA esposizione lorda che il libro puo' prendere:
-#   n_asset x frac x disaster_sl_pct x equity_totale
+#   book.leva_tetto() x disaster_sl_pct x equity_totale
 # tutto letto da config/live.json e src/live/book.py, MAI ridichiarato (P1). Prima del 10/09 la
 # formula viveva in scripts/live/usde_convert.py; il sorvegliante cuscino_watch la deriva da qui,
-# e usde_convert pure — due lettori, una formula.
-
-N_ASSET = 2
+# e usde_convert pure — due lettori, una formula. ⚠️ Fino al #28 la leva lorda era RIDICHIARATA
+# qui (N_ASSET fisso x WEIGHT, senza scala): a scala 1,25 dava 0,30 invece di 0,375. Ora la stessa
+# `book.leva_tetto` che usa anche il giornale (src/live/journal.py).
 
 
 def frazione_cuscino() -> tuple[float, str]:
@@ -117,8 +117,8 @@ def frazione_cuscino() -> tuple[float, str]:
     cfg = json.loads(CFG_PATH.read_text())
     sl = float(cfg["disaster_sl_pct"])
     from src.live import book as B          # lazy: book importa shadow, che importa questo modulo
-    frac = float(B.WEIGHT)
-    return N_ASSET * frac * sl, f"{N_ASSET} asset x frac {frac} x SL {sl:.0%}"
+    leva, come_leva = B.leva_tetto(cfg)
+    return leva * sl, f"{come_leva} x SL {sl:.0%}"
 
 
 def cuscino_richiesto_usd(equity_tot: float) -> tuple[float, str]:

src/live/venue_probe.py

@@ -28,19 +28,21 @@ col codice 11051, e' manutenzione; se non risponde affatto, e' il venue ma per u
 se la sonda stessa non parte, il verdetto e' IGNOTO — **«non vedo» non e' «va tutto bene»** (P5).
 La sonda parte SOLO quando qualcosa e' gia' fallito: sul percorso sano non costa niente.
 
-⚠️ **LA FINESTRA NON E' UNA PROVA.** `in_release_window()` dice soltanto che l'orologio e'
-compatibile con lo slot di release. Non declassa niente da sola: declassa solo cio' che la SONDA
-ha gia' riconosciuto come manutenzione, e solo dentro la durata ANNUNCIATA. Una manutenzione che
-sfora resta rumorosa (il 18/08 Deribit annuncio' 15-30 minuti e resto' giu' oltre l'ora), e un
-guasto vero di martedi' mattina resta rumoroso: altrimenti si sarebbe costruito il silenzio
-proprio nell'ora in cui e' piu' probabile che serva. E' la stessa logica di
-`venue_watch.MAINT_GRACE_HOURS` — *declassa dentro la tolleranza, RIALZA oltre* — applicata pero'
-a una domanda diversa (**questo giro** e' spiegato?) e quindi alla risoluzione del giro orario,
-non delle ore consecutive. Sono due politiche vicine e distinte: non accorparle senza misurare.
-
-REGOLA P9 — *un allarme massimo speso per un evento atteso e' un allarme che non verra' letto il
-giorno che e' vero.* E' il motivo per cui `atteso` esiste: declassa la GRAVITA', **non** registra
-di meno. L'episodio finisce nel log e nella notifica comunque.
+⚠️ **IL RAMO CHE DECLASSAVA E' MORTO, RIMOSSO DA `diagnose()` (#35 B11, 27/09/2026).**
+`in_release_window()` diceva se l'orologio cadeva nello slot di release Deribit (martedi'
+09:00-09:30 UTC); `diagnose()` la usava per declassare la GRAVITA' di una manutenzione gia'
+riconosciuta dalla SONDA (REGOLA P9 — *un allarme massimo speso per un evento atteso e' un
+allarme che non verra' letto il giorno che e' vero*). Ma l'UNICO chiamante di `diagnose()` e'
+`book_execute.py`, in cron al minuto **:47** di ogni ora (dal 25/08): la finestra 09:00-09:30 non
+contiene mai :47, quindi quel ramo non poteva MAI diventare vero — si leggeva come una
+declassazione attiva e non lo e' mai stata. Il campo `atteso` RESTA sul verdetto
+(`book_execute.py` lo scrive nel dict del report: cambiare il formato per una pulizia e' un costo
+che non vale la pena) ma e' hardcoded a False in `diagnose()`. `in_release_window()` e le
+costanti `RELEASE_*` restano invece nel modulo: le usa `test_book_cadenza.py`/`test_revisione.py`
+per verificare che i cron di book/review evitino lo stesso slot — uno scopo indipendente da
+`diagnose()` e ancora valido. Per declassare di nuovo un evento annunciato in `diagnose()` la
+strada e' legarlo alla DURATA dell'annuncio (`venue_news`, se e quando esiste), non all'orologio
+di un chiamante che non e' mai sincrono con lo slot.
 """
 from __future__ import annotations
 
@@ -53,7 +55,7 @@ import requests
 from src.live.venue_watch import is_maintenance   # P1: la firma 11051 vive LI', non qui
 
 # --- verdetti -----------------------------------------------------------------------------
-V_MANUTENZIONE = "VENUE_MANUTENZIONE"   # il venue dichiara 11051: atteso se dentro lo slot
+V_MANUTENZIONE = "VENUE_MANUTENZIONE"   # il venue dichiara 11051
 V_VENUE_GIU = "VENUE_GIU"               # il venue non risponde, e non dice manutenzione
 V_GATEWAY = "GATEWAY"                   # il venue sta bene -> il guasto e' NOSTRO (riparabile)
 V_IGNOTO = "IGNOTO"                     # la sonda non e' partita: "non vedo" (P5)
@@ -70,8 +72,12 @@ PROBE_TIMEOUT = 8.0
 # 30/06/2026 alle 09:00 UTC, "down for around 15-30 minutes").
 # ⚠️ M10: *due punti non fanno una tendenza*. Qui i punti sono quattro e c'e' una meccanica
 # dichiarata dal venue, ma resta una REGOLARITA' OSSERVATA, non un contratto: se Deribit sposta
-# lo slot, questa costante mente in silenzio. E' per questo che la finestra non decide da sola —
-# decide la sonda, e la finestra al massimo declassa.
+# lo slot, questa costante mente in silenzio.
+# ⚠️ #35 B11 (27/09/2026): questa finestra NON declassa piu' niente in `diagnose()` (vedi sotto):
+# l'unico chiamante gira in cron al minuto :47, che non ci cade mai dentro. La funzione e le
+# costanti restano perche' `test_book_cadenza.py`/`test_revisione.py` le usano per un controllo
+# diverso e ancora valido: che i cron di `book`/`review` cadano FUORI da questo slot (P1, la
+# stessa finestra, non una seconda dichiarata a mano).
 RELEASE_WEEKDAY = 1        # lunedi'=0 -> 1 = martedi'
 RELEASE_START_MIN = 9 * 60  # 09:00 UTC
 RELEASE_LEN_MIN = 30        # durata ANNUNCIATA. Oltre questa la manutenzione torna una notizia.
@@ -79,7 +85,8 @@ RELEASE_LEN_MIN = 30        # durata ANNUNCIATA. Oltre questa la manutenzione to
 
 @dataclass(frozen=True)
 class Diagnosi:
-    """Il verdetto (*cosa*), la spiegazione (*perche'*), e se l'evento era atteso (P9)."""
+    """Il verdetto (*cosa*) e la spiegazione (*perche'*). `atteso` resta nel formato ma e' sempre
+    False (#35 B11): l'unico chiamante non cade mai nella finestra che l'avrebbe declassato."""
     verdetto: str
     perche: str
     atteso: bool
@@ -87,7 +94,8 @@ class Diagnosi:
 
     @property
     def gravita(self) -> str:
-        """Emoji del titolo di notifica. `atteso` declassa, non silenzia (P9)."""
+        """Emoji del titolo di notifica. `atteso` e' sempre False (#35 B11): ramo morto lasciato
+        per compatibilita' di formato, non per una declassazione che avviene ancora."""
         if self.atteso:
             return "ℹ️"
         return {V_GATEWAY: "🛑", V_VENUE_GIU: "🛑", V_MANUTENZIONE: "⚠️"}.get(self.verdetto, "⚠️")
@@ -106,8 +114,9 @@ class Diagnosi:
 def in_release_window(now: datetime | None = None) -> bool:
     """L'orologio e' dentro lo slot di release Deribit (martedi' 09:00-09:30 UTC)?
 
-    PURA e testabile. Non e' una prova di niente da sola: vedi la nota in testa al modulo.
-    """
+    PURA e testabile. Non declassa piu' niente in `diagnose()` (#35 B11): resta per chi verifica
+    che un cron (book, review) sia schedulato fuori da questo slot (P1: la finestra si deriva da
+    qui, non si riscrive una seconda volta nel test)."""
     now = now or datetime.now(timezone.utc)
     now = now.astimezone(timezone.utc)
     if now.weekday() != RELEASE_WEEKDAY:
@@ -136,7 +145,6 @@ def probe_public(timeout: float = PROBE_TIMEOUT) -> tuple[bool | None, str]:
 
 
 def diagnose(errori_osservati: list[str] | None = None,
-             now: datetime | None = None,
              sonda=None) -> Diagnosi:
     """Classifica un guasto gia' avvenuto. `sonda` e' iniettabile: i test non toccano la rete.
 
@@ -169,7 +177,10 @@ def diagnose(errori_osservati: list[str] | None = None,
             verdetto = V_VENUE_GIU
             perche = "l'API pubblica Deribit non serve, e non dichiara manutenzione"
 
-    atteso = verdetto == V_MANUTENZIONE and in_release_window(now)
+    # #35 B11: qui c'era `verdetto == V_MANUTENZIONE and in_release_window(now)`. L'unico
+    # chiamante (book_execute.py, cron :47) non cade mai nella finestra 09:00-09:30 di martedi':
+    # il ramo non poteva mai diventare vero. Il campo resta (formato del report), sempre False.
+    atteso = False
     prova = "; ".join([*errori[:3], grezzo])[:500]
     return Diagnosi(verdetto=verdetto, perche=perche, atteso=atteso, prova=prova)
 

src/portfolio/sleeves.py

@@ -400,8 +400,14 @@ def deribit_book_sleeves() -> list[Sleeve]:
     e VRP01 (opzioni modellate). FULL Sharpe ~1.78 / HOLD ~1.17 / DD ~9% (research; SKH01 daily-step).
     Pensato per il deploy reale a basso capitale: stesso conto, stesso feed, ribilanciamento
     periodico (vedi src.portfolio.portfolio.rebalance_sim + scripts/portfolio/run_deribit_book.py).
-    TP01 e' gia' armato live; SKH01 e' il candidato 2a gamba (da validare codice d'esecuzione)."""
+    TP01 e' gia' armato live; SKH01 e' il candidato 2a gamba (da validare codice d'esecuzione).
+
+    I pesi si DERIVANO da src.live.book (P1): due letterali qui erano una seconda dichiarazione,
+    libera di divergere dal book vero (#35 B12). Import locale: book importa questo modulo
+    (_skyhook_positions) a livello di modulo -> un import in testa qui creerebbe un ciclo (come
+    fa usde.py con book)."""
+    from src.live import book as B    # lazy: vedi nota sul ciclo qui sopra
     return [
-        tp01_sleeve(weight=0.75),
-        skyhook_sleeve(weight=0.25),
+        tp01_sleeve(weight=B.W_TP01),
+        skyhook_sleeve(weight=B.W_SKH),
     ]

tests/test_analista.py

@@ -285,6 +285,39 @@ def test_la_testata_porta_i_numeri_che_si_leggono_dalla_notifica():
     t = A.testata(_voce_piena())
     assert "$637.36" in t and "+2.36" in t and "+39.30" in t
     assert "BTC $+116" in t and "leva 0.41x" in t and "1 fill" in t
+    # (#35 C8) il cron gira alle 00:30 sul giorno CHIUSO: non e' "oggi", e' "nel giorno".
+    assert "nel giorno" in t and "oggi" not in t
+
+
+def test_testata_con_movimento_mostra_trading_giorno_non_il_grezzo():
+    """(#35 C8) Un versamento di $1.399 non deve leggersi come P&L: con mov_giorno != 0 e
+    trading_giorno calcolabile, la cifra mostrata e' trading_giorno, non delta_equity."""
+    voce = _voce_piena()
+    voce["pnl"] = dict(voce["pnl"], delta_equity=1401.36, mov_giorno=1399.0, trading_giorno=2.36)
+    t = A.testata(voce)
+    assert "+2.36 nel giorno" in t
+    assert "di cui +1399.00 movimenti" in t
+    assert "+1401.36" not in t                    # il grezzo non compare piu' in testata
+
+
+def test_testata_con_movimento_ma_trading_giorno_none_resta_sul_grezzo():
+    """(#35 C8) Se trading_giorno non e' calcolabile (None) si resta su delta_equity, ma con
+    l'etichetta nuova, non "oggi"."""
+    voce = _voce_piena()
+    voce["pnl"] = dict(voce["pnl"], delta_equity=1401.36, mov_giorno=1399.0, trading_giorno=None)
+    t = A.testata(voce)
+    assert "+1401.36 nel giorno" in t
+    assert "di cui" not in t
+
+
+def test_testata_senza_movimento_mostra_il_grezzo_con_etichetta_nuova():
+    """(#35 C8) mov_giorno assente o 0: nessun movimento da scorporare, stesso trattamento di
+    journal.scorporo (bool())."""
+    voce = _voce_piena()
+    voce["pnl"] = dict(voce["pnl"], mov_giorno=0.0, trading_giorno=2.36)
+    t = A.testata(voce)
+    assert "+2.36 nel giorno" in t
+    assert "di cui" not in t
 
 
 def test_il_messaggio_sfugge_l_HTML_e_converte_il_grassetto():
@@ -312,6 +345,21 @@ def test_il_taglio_a_4096_si_DICHIARA():
     assert "tagliato" in m and "docs/journal/2026-08-23.md" in m
 
 
+def test_taglio_a_cavallo_di_entita_e_tag_resta_HTML_valido():
+    """(#35 C13) Il taglio e' sul GREZZO, prima di _b/html.escape e prima di **-><b>: un '&' o
+    un '**grassetto**' a cavallo del punto di taglio non deve produrre un'entita' spezzata
+    (es. '&am') ne' un <b> senza chiusura -- Telegram rifiuta l'INTERO messaggio altrimenti
+    ("can't parse entities"). Il pattern si ripete per tutta la lunghezza, cosi' il punto di
+    taglio (qualunque sia) cade sempre dentro o a cavallo di un '&' o di '**...**'."""
+    pattern = "parola & altra parola **in grassetto** ancora testo "
+    analisi = pattern * 200                          # ben oltre qualunque budget qui
+    m = A.per_telegram(_voce_piena(), analisi, "ok", [])
+    assert "tagliato" in m
+    assert re.search(r"&(?!amp;|lt;|gt;)", m) is None    # nessuna entita' spezzata o '&' nudo
+    assert m.count("<b>") == m.count("</b>")              # ogni <b> aperto e' chiuso
+    assert m.count("<i>") == m.count("</i>")
+
+
 def test_un_messaggio_normale_non_viene_toccato():
     m = A.per_telegram(_voce_piena(), "Analisi breve.", "ok", [])
     assert "tagliato" not in m and m.endswith("Analisi breve.")

tests/test_book_live.py

@@ -150,6 +150,27 @@ def test_weights_match_backtest_sleeves():
     assert abs((W_TP01 + W_SKH) - 1.0) < 1e-12
 
 
+def test_deribit_book_sleeves_deriva_i_pesi_da_book_non_li_ridichiara():
+    """#35 B12 — P1: stesso pattern di test_r0909_libro_nei_crolli.test_P1_i_pesi_sono_QUELLI_del_book_live_non_una_copia.
+    `test_weights_match_backtest_sleeves` sopra passerebbe anche con due letterali per coincidenza
+    di valore (0.75/0.25 riscritti a mano): qui si verifica l'IDENTITA' con B.W_TP01/B.W_SKH e
+    che il letterale non sia piu' nella fonte di `deribit_book_sleeves`."""
+    import inspect
+    import re
+
+    import src.live.book as B
+    from src.portfolio.sleeves import deribit_book_sleeves
+
+    pesi = {s.name.split("_")[0]: s.weight for s in deribit_book_sleeves()}
+    assert pesi["TP01"] is B.W_TP01 or pesi["TP01"] == B.W_TP01
+    assert pesi["SKH01"] is B.W_SKH or pesi["SKH01"] == B.W_SKH
+
+    src = inspect.getsource(deribit_book_sleeves)
+    assert "from src.live import book" in src, "i pesi devono derivare da src.live.book (P1)"
+    assert not re.search(r"weight\s*=\s*0\.75", src), "peso TP01 ancora cablato"
+    assert not re.search(r"weight\s*=\s*0\.25", src), "peso SKH01 ancora cablato"
+
+
 # ---------------------------------------------------------------------------
 # Costruzione ordine: side, reduce_only, flip, close, soglia minima.
 # ---------------------------------------------------------------------------

tests/test_cuscino_watch.py

@@ -155,7 +155,15 @@ def test_un_tentativo_ogni_sei_ore():
     assert W.puo_riconvertire([{"ts": now - 1000, "riconversione": None}], now, True, 1.0, 0.99)[0]
 
 
+def _finto_lettore_quota(valori):
+    """Ritorna i valori dati in ordine, uno per chiamata (per monkeypatch di _leggi_quota_usde)."""
+    it = iter(valori)
+    return lambda: next(it)
+
+
 def test_riconverti_lancia_lo_script_vero_con_esegui(monkeypatch):
+    # rc==0: "ok" a prescindere dalla quota (lo script vero ha gia' le sue guardie, P15) — la
+    # quota qui non si muove, per mostrare che il caso "ok" non dipende dalla rilettura.
     chiamate = []
 
     class R:
@@ -166,10 +174,61 @@ def test_riconverti_lancia_lo_script_vero_con_esegui(monkeypatch):
         return R()
 
     monkeypatch.setattr(W.subprocess, "run", finto_run)
+    monkeypatch.setattr(W, "_leggi_quota_usde", _finto_lettore_quota([0.67, 0.67]))
     e = W.riconverti(0.67)
-    assert e["ok"] and chiamate and chiamate[0][1].endswith("usde_convert.py")
+    assert e["ok"] and e["esito"] == "ok" and chiamate and chiamate[0][1].endswith("usde_convert.py")
     assert "--esegui" in chiamate[0] and "0.6700" in chiamate[0]
     assert e["coda"][-1].startswith("  ESITO")
+    assert e["quota_prima"] == 0.67 and e["quota_dopo"] == 0.67
+
+
+def test_riconverti_rc_diverso_da_zero_con_quota_mossa_e_parziale(monkeypatch):
+    # #35 B10: rc!=0 (es. ordine parzialmente fillato poi rifiutato) ma la quota si e' mossa
+    # per davvero -> "parziale", non "fallita": la vendita/acquisto e' avvenuto, solo non finito.
+    class R:
+        returncode, stdout, stderr = 1, "", "book illeggibile a meta' piano\n"
+
+    monkeypatch.setattr(W.subprocess, "run", lambda cmd, **kw: R())
+    monkeypatch.setattr(W, "_leggi_quota_usde", _finto_lettore_quota([0.75, 0.70]))
+    e = W.riconverti(0.64)
+    assert e["esito"] == "parziale" and e["ok"] is False
+    assert e["quota_prima"] == 0.75 and e["quota_dopo"] == 0.70
+
+
+def test_riconverti_rc_diverso_da_zero_con_quota_ferma_e_fallita(monkeypatch):
+    class R:
+        returncode, stdout, stderr = 1, "", "prezzo fuori banda: niente ordini\n"
+
+    monkeypatch.setattr(W.subprocess, "run", lambda cmd, **kw: R())
+    monkeypatch.setattr(W, "_leggi_quota_usde", _finto_lettore_quota([0.75, 0.751]))
+    e = W.riconverti(0.64)
+    assert e["esito"] == "fallita" and e["ok"] is False
+
+
+def test_riconverti_timeout_con_quota_mossa_e_parziale(monkeypatch):
+    # #35 B10: il subprocess viene ucciso a TIMEOUT_CONVERT_S ma alcuni ordini erano gia' fillati
+    # prima del kill -> oggi (ok=rc==0) diventava "FALLITA", corretto e' "parziale".
+    def finto_run(cmd, **kw):
+        raise W.subprocess.TimeoutExpired(cmd, kw.get("timeout"))
+
+    monkeypatch.setattr(W.subprocess, "run", finto_run)
+    monkeypatch.setattr(W, "_leggi_quota_usde", _finto_lettore_quota([0.80, 0.72]))
+    e = W.riconverti(0.64)
+    assert e["rc"] is None and e["esito"] == "parziale" and e["ok"] is False
+    assert e["coda"][0].startswith("timeout dopo")
+
+
+def test_riconverti_rilettura_dopo_fallita_e_fallita_esplicita(monkeypatch):
+    # se non si riesce a rileggere la quota dopo, non si puo' dire "parziale": si dichiara
+    # l'incertezza invece di dedurre un movimento zero che non si e' verificato.
+    class R:
+        returncode, stdout, stderr = 1, "", "errore\n"
+
+    monkeypatch.setattr(W.subprocess, "run", lambda cmd, **kw: R())
+    monkeypatch.setattr(W, "_leggi_quota_usde", _finto_lettore_quota([0.75, None]))
+    e = W.riconverti(0.64)
+    assert e["esito"] == "fallita (quota dopo non leggibile)" and e["ok"] is False
+    assert e["quota_dopo"] is None
 
 
 def test_il_monitor_e_registrato_in_monitor_health():

tests/test_fee_watch.py

@@ -204,7 +204,10 @@ def test_invio_riuscito_salva_lo_stato(monkeypatch):
     salvato, inviato = _prepara(monkeypatch, _RIS_AZIONE, esito_invio=True)
     FW.main()
     assert len(inviato) == 1
-    assert len(salvato) == 1 and salvato[0] == _RIS_AZIONE["current"]
+    # (#37 B9) lo stato salvato porta anche il marcatore "divergenza_detta" (qui False: la
+    # fixture non ha divergenze), sibling delle fee per-strumento, non piu' solo `current`.
+    assert len(salvato) == 1
+    assert salvato[0] == {**_RIS_AZIONE["current"], "divergenza_detta": False}
 
 
 def test_nessun_invio_dovuto_salva_comunque(monkeypatch):
@@ -213,3 +216,96 @@ def test_nessun_invio_dovuto_salva_comunque(monkeypatch):
     FW.main()
     assert inviato == [], "senza changes e livello OK non deve tentare l'invio"
     assert len(salvato) == 1
+
+
+# ===========================================================================
+# issue #37 (B9) — realized_fee_bps() SEMPRE calcolato (non solo nel blocco di stampa) e
+# divergenza dal tier pubblico notificata UNA volta per episodio.
+# ===========================================================================
+def test_run_calcola_sempre_la_fee_realizzata(monkeypatch):
+    """Prima del #37 la fee realizzata girava solo dentro `if not quiet or interessante`: in
+    cron (--quiet) non girava mai. Ora e' `run()` a calcolarla, sempre."""
+    monkeypatch.setattr(FW, "fetch_fees",
+                         lambda ins: dict(instrument=ins, taker_bps=5.0, maker_bps=0.0, liq_bps=75.0))
+    monkeypatch.setattr(FW, "load_state", lambda: {})
+    ins = FW.INSTRUMENTS[0]
+    monkeypatch.setattr(FW, "realized_fee_bps", lambda limit=20: {ins: 7.5})
+    r = FW.run()
+    assert r["realized"] == {ins: 7.5}
+    assert len(r["divergenze"]) == 1 and ins in r["divergenze"][0]
+    assert r["divergenza_gia_detta"] is False
+
+
+def test_run_nessun_trade_e_None_non_zero(monkeypatch):
+    monkeypatch.setattr(FW, "fetch_fees",
+                         lambda ins: dict(instrument=ins, taker_bps=5.0, maker_bps=0.0, liq_bps=75.0))
+    monkeypatch.setattr(FW, "load_state", lambda: {})
+    monkeypatch.setattr(FW, "realized_fee_bps", lambda limit=20: {})
+    r = FW.run()
+    assert r["realized"] is None
+    assert r["divergenze"] == []
+
+
+def test_run_sotto_soglia_non_e_divergenza(monkeypatch):
+    monkeypatch.setattr(FW, "fetch_fees",
+                         lambda ins: dict(instrument=ins, taker_bps=5.0, maker_bps=0.0, liq_bps=75.0))
+    monkeypatch.setattr(FW, "load_state", lambda: {})
+    ins = FW.INSTRUMENTS[0]
+    monkeypatch.setattr(FW, "realized_fee_bps", lambda limit=20: {ins: 5.3})  # +0.3 < 1.0
+    r = FW.run()
+    assert r["divergenze"] == []
+
+
+_RIS_DIVERGE = dict(
+    current={"BTC_USDC-PERPETUAL": dict(taker_bps=5.0, maker_bps=0.0, liq_bps=75.0)},
+    changes=[], worst_taker_bps=5.0, level="OK", reason="ok", first_read=False,
+    realized={"BTC_USDC-PERPETUAL": 7.5},
+    divergenze=["BTC_USDC-PERPETUAL: realizzata 7.50 bps vs tier pubblico 5.00 (+2.50)"],
+    divergenza_gia_detta=False,
+)
+_RIS_DIVERGE_RIPETUTA = dict(_RIS_DIVERGE, divergenza_gia_detta=True)
+_RIS_DIVERGE_RIENTRATA = dict(
+    current={"BTC_USDC-PERPETUAL": dict(taker_bps=5.0, maker_bps=0.0, liq_bps=75.0)},
+    changes=[], worst_taker_bps=5.0, level="OK", reason="ok", first_read=False,
+    realized={"BTC_USDC-PERPETUAL": 5.3}, divergenze=[], divergenza_gia_detta=True,
+)
+
+
+def _prepara_b9(monkeypatch, risultato, esito_invio):
+    """Come `_prepara`, ma senza `changes`/livello AZIONE: isola il notify di B9 da quello di
+    B8 (qui non richiesto, dato che tutte le fixture hanno level='OK' e changes=[])."""
+    salvato, inviato = [], []
+    monkeypatch.setattr(FW, "run", lambda: risultato)
+    monkeypatch.setattr(FW, "save_state", lambda cur: salvato.append(cur))
+    monkeypatch.setattr(FW, "notify", lambda *a, **k: inviato.append((a, k)) or esito_invio)
+    monkeypatch.setattr(FW, "registra_giro", lambda *a, **k: None)
+    return salvato, inviato
+
+
+def test_divergenza_prima_volta_notifica_e_marca(monkeypatch):
+    salvato, inviato = _prepara_b9(monkeypatch, _RIS_DIVERGE, esito_invio=True)
+    FW.main()
+    assert len(inviato) == 1
+    assert salvato[-1]["divergenza_detta"] is True
+
+
+def test_divergenza_stesso_episodio_non_ripete(monkeypatch):
+    salvato, inviato = _prepara_b9(monkeypatch, _RIS_DIVERGE_RIPETUTA, esito_invio=True)
+    FW.main()
+    assert inviato == [], "gia' detta in questo episodio: nessun secondo invio"
+    assert salvato[-1]["divergenza_detta"] is True
+
+
+def test_divergenza_notify_fallita_non_marca(monkeypatch):
+    salvato, inviato = _prepara_b9(monkeypatch, _RIS_DIVERGE, esito_invio=False)
+    FW.main()
+    assert len(inviato) == 1
+    assert salvato[-1]["divergenza_detta"] is False, \
+        "invio fallito -> marcatore non scritto, il giro dopo ritenta"
+
+
+def test_divergenza_rientrata_azzera_il_marcatore(monkeypatch):
+    salvato, inviato = _prepara_b9(monkeypatch, _RIS_DIVERGE_RIENTRATA, esito_invio=False)
+    FW.main()
+    assert inviato == [], "sotto soglia: nessun invio da tentare"
+    assert salvato[-1]["divergenza_detta"] is False

tests/test_leva_tetto.py

@@ -0,0 +1,140 @@
+"""book.leva_tetto — la leva lorda massima del libro, in UNA sola formula (#28).
+
+Prima della correzione la leva lorda tornava con tre formule diverse: `usde.frazione_cuscino`
+(N_ASSET fisso, senza scala), la regola `leva` del giornale (senza scala) e
+`journal.movimenti_capitale` (con scala, ma ridichiarata). A scala 1,25 il cuscino USDC
+risultava 0,30 dell'equity invece di 0,375 — il caso che T2 riproduce.
+"""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from src.live import book as B  # noqa: E402
+from src.live import journal as J  # noqa: E402
+from src.live import tradesdb as T  # noqa: E402
+from src.live import usde as U  # noqa: E402
+
+
+def _cfg(**kw) -> dict:
+    base = dict(max_notional_per_asset_frac=0.5, disaster_sl_pct=0.3)
+    base.update(kw)
+    return base
+
+
+# ---------------------------------------------------------------- T1: scala assente
+
+def test_T1_scala_assente_e_frac_uguale_al_weight():
+    leva, come = B.leva_tetto(_cfg())
+    assert leva == pytest.approx(len(B.ASSETS) * B.WEIGHT)
+    assert "scala 1.0" in come
+
+
+# ---------------------------------------------------------------- T2: il caso dell'issue
+
+def test_T2_scala_1_25_alza_la_leva_e_il_cuscino_usde(tmp_path, monkeypatch):
+    """A scala 1,25 il cuscino USDC deve valere 0,375 dell'equity, non 0,30 (#28)."""
+    cfg = _cfg(book_scale_k=1.25)
+    leva, _ = B.leva_tetto(cfg)
+    assert leva == pytest.approx(1.25)
+
+    # frazione_cuscino legge da CFG_PATH: si monkeypatcha il file, come fanno gli altri test
+    # su usde/_scala (grep "CFG_PATH\|CONFIG" tests/).
+    cpath = tmp_path / "live.json"
+    cpath.write_text(json.dumps(cfg))
+    monkeypatch.setattr(U, "CFG_PATH", cpath)
+    f, come = U.frazione_cuscino()
+    assert f == pytest.approx(0.375)
+    assert "SL 30%" in come
+
+
+# ---------------------------------------------------------------- T3: fuori scaletta non solleva
+
+def test_T3_scala_fuori_scaletta_non_solleva_e_torna_il_tetto_di_codice():
+    """cuscino_watch non deve morire proprio quando la config e' sbagliata (verifica fable B5)."""
+    leva, come = B.leva_tetto(_cfg(book_scale_k=1.1))
+    assert leva == pytest.approx(B.LEVA_LORDA_MAX)
+    assert "tetto di codice" in come
+
+
+# ---------------------------------------------------------------- T4: frac assente
+
+def test_T4_frac_assente_usa_weight():
+    cfg = dict(disaster_sl_pct=0.3)          # niente max_notional_per_asset_frac
+    leva, come = B.leva_tetto(cfg)
+    assert leva == pytest.approx(len(B.ASSETS) * B.WEIGHT)
+    assert f"frac {B.WEIGHT}" in come
+
+
+# ---------------------------------------------------------------- T5: il tronco al tetto di codice
+
+def test_T5_leva_oltre_il_tetto_di_codice_viene_troncata(monkeypatch):
+    """Verifica diretta del troncamento. Con WEIGHT alzato il prodotto supera LEVA_LORDA_MAX pur
+    restando `_scala` silenziosa (frac ASSENTE dalla config -> il suo controllo sul prodotto non
+    scatta, perche' guarda solo `max_notional_per_asset_frac` quando e' dichiarato)."""
+    monkeypatch.setattr(B, "WEIGHT", 0.9)
+    leva, come = B.leva_tetto(dict(book_scale_k=1.25))   # frac assente -> usa WEIGHT = 0.9
+    assert leva == pytest.approx(B.LEVA_LORDA_MAX)
+    assert "troncato" in come
+
+
+# ---------------------------------------------------------------- T6: il giornale usa la stessa leva_tetto
+
+@pytest.fixture()
+def db(tmp_path):
+    con = T.connect(tmp_path / "t.db")
+    yield con
+    con.close()
+
+
+def _voce(mercato=None, libro=None, pnl=None, salute=None, giorno="2026-08-21"):
+    return dict(giorno=giorno, ts_scritto=T.ora(), mercato=mercato or {},
+                libro=libro or {"errore": "nessun giro", "giri": 0}, pnl=pnl or {},
+                salute=salute or dict(giri_book=24, feed_skh_min=0, atteso_giri=24,
+                                      giorno_in_corso=False, giri_mancanti=0))
+
+
+def _libro(**asset):
+    a = {k: dict(dict(tp_frac=0.0, skh_sign=0, skh_entry=None, target=0.0, posizione=0.0,
+                      azione="HOLD (a target)"), **v) for k, v in asset.items()}
+    lordo = sum(abs(v["posizione"]) for v in a.values())
+    return dict(giri=24, ultimo_giro="2026-08-21T23:07:01+00:00", equity=600.0, barra="2026-08-21",
+                feed_skh_min=0, nozionale_lordo=lordo, leva_lorda=lordo / 600.0, asset=a)
+
+
+def _pnl(**kw):
+    base = dict(equity_inizio=600.0, equity_fine=600.0, delta_equity=0.0, realizzato_lordo=0.0,
+                realizzato_netto=0.0, roundtrip_chiusi=0, fill=0, fee=0.0,
+                cumulato_da_arming=0.0, letture_equity=24)
+    return {**base, **kw}
+
+
+def test_T6_regola_leva_margine_positivo_a_scala_1_25(db, monkeypatch, tmp_path):
+    """lv=1.2 su un tetto di 1.25x (frac 0.5, scala 1.25): margine positivo, non negativo."""
+    (tmp_path / "config").mkdir()
+    (tmp_path / "config" / "live.json").write_text(
+        json.dumps(dict(max_notional_per_asset_frac=0.5, book_scale_k=1.25)))
+    monkeypatch.setattr(J, "PROJECT_ROOT", tmp_path)
+    lb = _libro(BTC=dict(posizione=360.0), ETH=dict(posizione=360.0))  # 720/600 = 1.2x
+    assert lb["leva_lorda"] == pytest.approx(1.2)
+    r = J.lettura(db, _voce(libro=lb, pnl=_pnl()))
+    riga = next(x for x in r if x["id"] == "leva")
+    assert "1.25x" in riga["testo"] and "margine 0.05x" in riga["testo"]
+
+
+# ---------------------------------------------------------------- A7 (#35): min_order_usd da config
+
+def test_A7_build_book_order_rispetta_min_usd_di_config():
+    """book_report legge min_order_usd da config (default 5.0 se assente, come book_execute.py).
+    Qui si verifica direttamente build_book_order, che e' il punto dove il valore viene usato:
+    con min_usd=20 un delta di $10 non genera ordine, con min_usd=5 (il default di oggi) si."""
+    order_stretto = B.build_book_order("BTC_USDC-PERPETUAL", 100.0, 90.0, 50000.0, min_usd=20.0)
+    assert order_stretto is None                  # |delta|=10 < 20: sotto soglia col tetto alzato
+    order_default = B.build_book_order("BTC_USDC-PERPETUAL", 100.0, 90.0, 50000.0, min_usd=5.0)
+    assert order_default is not None               # |delta|=10 >= 5: lo stesso caso, oggi, genera ordine

tests/test_monitor_health_orari.py

@@ -0,0 +1,140 @@
+"""Test del giro ORARIO di monitor_health (#30).
+
+I sorveglianti a max_age<=6h (cuscino_watch, venue_watch, collect_chain) girano alle 00:30 col
+report giornaliero e basta: un guasto si scopre fino a 24h dopo. Qui si guarda ogni ora, ma si
+notifica solo cio' che e' NUOVO rispetto al marcatore "gia' detto" — altrimenti un monitor fermo
+manderebbe lo stesso Telegram ogni ora per giorni.
+
+Le funzioni PURE (filtro per max_age, chiave dell'alert, chi e' nuovo, il marcatore aggiornato)
+stanno in src/live/monitor_health.py e si testano su righe finte, senza toccare disco o rete.
+L'orchestrazione (leggere/scrivere il marcatore, chiamare notify) sta nello script e si testa qui
+sotto con monkeypatch — NIENTE Telegram vero."""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+sys.path.insert(0, str(ROOT / "scripts" / "live"))
+
+from src.live.monitor_health import (  # noqa: E402
+    MonitorSpec,
+    alert_nuovi,
+    chiave_alert,
+    marcatore_aggiornato,
+    monitors_orari,
+)
+
+import monitor_health as MH  # noqa: E402  (scripts/live/monitor_health.py)
+
+# righe finte, nella forma che assess()/check_all() restituiscono davvero
+ROW_FERMO = dict(name="venue_watch", gate="tripwire di venue orario", n_bars=3, age_h=4.0,
+                  coverage=0.9, expected=3, status="FERMO", why="ultima barra 4h fa (limite 3h)")
+ROW_OK = dict(name="venue_watch", gate="tripwire di venue orario", n_bars=10, age_h=0.2,
+              coverage=1.0, expected=10, status="OK", why="")
+ROW_NUOVO = dict(name="venue_watch", gate=None, n_bars=1, age_h=0.1, coverage=None,
+                  expected=None, status="NUOVO", why="1 barra: non c'e' abbastanza serie")
+
+
+# ===========================================================================
+# funzioni PURE — filtro per max_age (src/live/monitor_health.py)
+# ===========================================================================
+def test_il_filtro_deriva_dalla_tabella_non_da_una_lista_di_nomi():
+    specs = (
+        MonitorSpec("a", "a", "runs.jsonl", 1.0, max_age_h=3.0),
+        MonitorSpec("b", "b", "runs.jsonl", 1.0, max_age_h=6.0),      # al bordo: dentro
+        MonitorSpec("c", "c", "runs.jsonl", 24.0, max_age_h=30.0),
+    )
+    assert {m.name for m in monitors_orari(specs)} == {"a", "b"}
+
+
+def test_i_tre_orari_di_oggi_sono_quelli_noti():
+    """Se la tabella cambia questo test lo dice: e' l'ancora fra il registro e il flag."""
+    assert {m.name for m in monitors_orari()} == {"collect_chain", "cuscino_watch", "venue_watch"}
+
+
+# ===========================================================================
+# funzioni PURE — chiave, nuovi, marcatore
+# ===========================================================================
+def test_chiave_e_nome_piu_stato_non_il_testo():
+    """Il testo (`why`) contiene l'eta', che cambia ogni ora anche se il guasto e' lo stesso."""
+    assert chiave_alert(ROW_FERMO) == "venue_watch:FERMO"
+    piu_vecchio = {**ROW_FERMO, "age_h": 9.0, "why": "ultima barra 9h fa (limite 3h)"}
+    assert chiave_alert(piu_vecchio) == chiave_alert(ROW_FERMO)
+
+
+def test_alert_nuovo_se_la_chiave_non_e_nel_marcatore():
+    assert alert_nuovi([ROW_FERMO], set()) == [ROW_FERMO]
+
+
+def test_lo_stesso_alert_al_secondo_giro_non_e_piu_nuovo():
+    assert alert_nuovi([ROW_FERMO], {"venue_watch:FERMO"}) == []
+
+
+def test_nuovo_non_e_un_alert_da_notificare():
+    assert alert_nuovi([ROW_NUOVO], set()) == []
+
+
+def test_il_marcatore_aggiunge_la_chiave_di_un_alert():
+    assert marcatore_aggiornato([ROW_FERMO], set()) == {"venue_watch:FERMO"}
+
+
+def test_il_ritorno_a_ok_toglie_la_chiave_dal_marcatore():
+    """Cosi' un nuovo guasto sullo stesso monitor riparte con un nuovo avviso."""
+    assert marcatore_aggiornato([ROW_OK], {"venue_watch:FERMO"}) == set()
+
+
+# ===========================================================================
+# orchestrazione — scripts/live/monitor_health.py main_orari(), NIENTE Telegram
+# ===========================================================================
+def test_notify_fallito_non_scrive_il_marcatore(tmp_path, monkeypatch):
+    """Stessa regola di fee_watch #29: niente 'gia' detto' su un invio fallito, altrimenti il
+    prossimo giro non ritenterebbe."""
+    detto = tmp_path / "detto.json"
+    monkeypatch.setattr(MH, "DETTO", detto)
+    monkeypatch.setattr(MH, "check_all", lambda root, monitors=None: [ROW_FERMO])
+    chiamate = []
+    monkeypatch.setattr(MH, "notify", lambda *a, **k: (chiamate.append((a, k)), False)[1])
+    monkeypatch.setattr(sys, "argv", ["monitor_health.py", "--quiet", "--solo-orari"])
+
+    rc = MH.main_orari()
+
+    assert rc == 2                 # c'e' un alert
+    assert len(chiamate) == 1      # e si e' provato a notificarlo
+    assert not detto.exists()      # ma l'invio e' fallito: nessun marcatore
+
+
+def test_notify_riuscito_scrive_il_marcatore_e_non_rinotifica(tmp_path, monkeypatch):
+    detto = tmp_path / "detto.json"
+    monkeypatch.setattr(MH, "DETTO", detto)
+    monkeypatch.setattr(MH, "check_all", lambda root, monitors=None: [ROW_FERMO])
+    monkeypatch.setattr(sys, "argv", ["monitor_health.py", "--quiet", "--solo-orari"])
+
+    monkeypatch.setattr(MH, "notify", lambda *a, **k: True)
+    rc1 = MH.main_orari()
+    assert rc1 == 2
+    assert json.loads(detto.read_text())["detti"] == ["venue_watch:FERMO"]
+
+    chiamate = []
+    monkeypatch.setattr(MH, "notify", lambda *a, **k: (chiamate.append(1), True)[1])
+    rc2 = MH.main_orari()
+    assert rc2 == 2             # l'alert c'e' ancora (anche se gia' detto: exit 2 comunque)
+    assert chiamate == []       # ma non si rimanda lo stesso Telegram
+
+
+def test_tornato_ok_il_giro_dopo_pulisce_il_marcatore(tmp_path, monkeypatch):
+    detto = tmp_path / "detto.json"
+    detto.write_text(json.dumps({"detti": ["venue_watch:FERMO"], "ts": "2026-09-27T00:00:00Z"}))
+    monkeypatch.setattr(MH, "DETTO", detto)
+    monkeypatch.setattr(MH, "check_all", lambda root, monitors=None: [ROW_OK])
+    monkeypatch.setattr(sys, "argv", ["monitor_health.py", "--quiet", "--solo-orari"])
+    chiamate = []
+    monkeypatch.setattr(MH, "notify", lambda *a, **k: (chiamate.append(1), True)[1])
+
+    rc = MH.main_orari()
+
+    assert rc == 0
+    assert chiamate == []        # OK non notifica mai
+    assert json.loads(detto.read_text())["detti"] == []

tests/test_revisione.py

@@ -68,14 +68,65 @@ def test_i_sorveglianti_vengono_prima_delle_voci_grandi(tmp_path):
 
 
 def test_il_troncamento_e_dichiarato(tmp_path, monkeypatch):
+    """CLAUDE.md e' la prima voce e ha un tetto SUO (MAX_CHARS_CLAUDE_MD, #37 C12): si monkeypatcha
+    quello, non il MAX_CHARS_VOCE generico, che non lo tocca piu'."""
     r = _radice(tmp_path)
     (r / "CLAUDE.md").write_text("x" * 1000)
-    monkeypatch.setattr(R, "MAX_CHARS_VOCE", 100)
+    monkeypatch.setattr(R, "MAX_CHARS_CLAUDE_MD", 100)
     n, t, tr = R.materiale(r, ADESSO)[0]
     assert tr and "[… troncato a 100 caratteri su 1,000]" in t
     assert "[TRONCATO]" in R.costruisci_prompt([(n, t, tr)], ADESSO)
 
 
+def test_claude_md_ha_un_tetto_suo_distinto_dal_generico(tmp_path, monkeypatch):
+    """#37 C12: CLAUDE.md sfiora MAX_CHARS_VOCE con un margine minimo — deve avere un tetto suo
+    piu' largo, cosi' un piccolo aggiornamento non lo tronca."""
+    r = _radice(tmp_path)
+    n_chars = R.MAX_CHARS_VOCE + 1_000     # sopra il generico, ma sotto MAX_CHARS_CLAUDE_MD
+    assert n_chars < R.MAX_CHARS_CLAUDE_MD
+    (r / "CLAUDE.md").write_text("x" * n_chars)
+    n, t, tr = R.materiale(r, ADESSO)[0]
+    assert n.startswith("CLAUDE.md")
+    assert not tr, "sotto il tetto SUO (150k): non deve troncare, anche se supera i 90k generici"
+
+
+def test_voce_vicina_al_tetto_e_avvisata_nel_prompt_e_nel_rapporto(tmp_path, monkeypatch):
+    """#37 C12: al 95% del tetto (non ancora troncata) l'avviso entra nel testo, e da li' arriva
+    sia al prompt del revisore sia alla tabella del rapporto — come [TRONCATO] oggi."""
+    r = _radice(tmp_path)
+    monkeypatch.setattr(R, "MAX_CHARS_CLAUDE_MD", 10_000)
+    (r / "CLAUDE.md").write_text("x" * 9_600)             # 96% di 10.000, margine largo per la nota
+    n, t, tr = R.materiale(r, ADESSO)[0]
+    assert not tr, "non e' troncata: e' vicina"
+    assert R._MARCA_VICINO_TETTO in t
+    assert "[VICINO AL TETTO]" in R.costruisci_prompt([(n, t, tr)], ADESSO)
+    rap = R.rapporto("revisione", None, "ok", [], "claude-fable-5-1", ADESSO, [(n, t, tr)])
+    assert "| quasi |" in rap
+
+
+def test_voce_lontana_dal_tetto_non_e_avvisata(tmp_path):
+    voci = R.materiale(_radice(tmp_path), ADESSO)
+    claude_md = [t for n, t, tr in voci if n.startswith("CLAUDE.md")][0]
+    assert R._MARCA_VICINO_TETTO not in claude_md
+
+
+def test_il_totale_resta_sotto_il_tetto_anche_se_la_somma_dei_tetti_lo_supera(tmp_path, monkeypatch):
+    """#37 C12: CLAUDE.md (150k) + il resto delle voci (90k l'una) puo' superare facilmente i 400k
+    di MAX_CHARS_TOTALE in somma — il tetto totale e' applicato A PARTE, a valle, e resta il vero
+    vincolo: non deve MAI sfondare, qualunque sia il tetto di ogni singola voce."""
+    r = _radice(tmp_path)
+    (r / "CLAUDE.md").write_text("x" * R.MAX_CHARS_CLAUDE_MD)      # al SUO tetto: 150k
+    j = r / "docs" / "journal"
+    for i in range(1, 10):                                          # sovrascrive i giornali della finestra
+        g = (ADESSO - timedelta(days=i)).date().isoformat()
+        (j / f"{g}.md").write_text("y" * R.MAX_CHARS_VOCE)           # ognuno al SUO tetto: 90k
+    voci = R.materiale(r, ADESSO)
+    tot = sum(len(t) for _, t, _ in voci)
+    somma_tetti = sum(R._cap_voce(n) for n, _, _ in voci)
+    assert somma_tetti > R.MAX_CHARS_TOTALE, "il caso e' quello che sfonderebbe se il totale non fosse a parte"
+    assert tot <= R.MAX_CHARS_TOTALE
+
+
 def test_il_prompt_porta_le_regole_e_la_data(tmp_path):
     p = R.costruisci_prompt(R.materiale(_radice(tmp_path), ADESSO), ADESSO)
     assert "DATA DELLA REVISIONE: 2026-09-14 06:15Z" in p
@@ -132,6 +183,18 @@ def test_telegram_porta_la_sintesi_e_dichiara_il_taglio():
     assert "[… sintesi tagliata" in lunga and len(lunga) < 4096
 
 
+def test_taglio_a_cavallo_di_una_entita_resta_HTML_valido():
+    """(#35 C13) Un '&' esattamente al taglio non deve spezzarsi in un'entita' a meta' (es.
+    '&am'): il taglio e' sul GREZZO, prima di html.escape, quindi il carattere e' incluso o
+    escluso per intero — mai spezzato a meta' entita'."""
+    sintesi = "a" * (R.LIMITE_TELEGRAM - 1) + "&resto che verra' tagliato via di sicuro"
+    tg = R.per_telegram(_revisione(sintesi=sintesi), None, "ok", ADESSO, Path("p.md"))
+    assert "[… sintesi tagliata" in tg
+    corpo = tg.split("\n\n", 1)[1].split("\n[…", 1)[0]
+    assert corpo == "a" * (R.LIMITE_TELEGRAM - 1) + "&amp;"   # il '&' e' incluso PER INTERO
+    assert "&am;" not in tg and "resto che verra'" not in tg  # niente entita' a meta', niente coda
+
+
 # ---------------------------------------------------------------- orchestrazione: cosa scrive, cosa no
 
 def _albero(root: Path) -> set[str]:

tests/test_telegram_daily.py

@@ -7,6 +7,7 @@ e non invia niente in dry-run.
 from __future__ import annotations
 
 import sys
+from datetime import date
 from pathlib import Path
 
 import pytest
@@ -21,8 +22,33 @@ telegram_daily = pytest.importorskip("telegram_daily")
 def test_build_non_solleva_e_ha_le_sezioni():
     txt = telegram_daily.build()
     assert isinstance(txt, str) and len(txt) > 50
-    for sezione in ("Conto", "Perche' non opera", "Ordini", "Monitor in osservazione"):
+    for sezione in ("Conto", "Ordini", "Monitor in osservazione"):
         assert sezione in txt, f"sezione mancante: {sezione}"
+    # (#35 C14) l'header sul trend e' condizionale: esattamente uno dei due, non piu' fisso.
+    assert ("Perche' non opera" in txt) != ("Stato trend" in txt)
+
+
+def test_header_perche_non_opera_solo_se_il_libro_e_flat(monkeypatch):
+    """(#35 C14) L'header fisso mentiva quando il libro NON e' fermo (posizioni o target != 0):
+    con flat=True resta "Perche' non opera"."""
+    monkeypatch.setattr(telegram_daily, "book_state", lambda: (["  equity $1.00"], True))
+    assert "Perche' non opera" in telegram_daily.build()
+
+
+def test_header_neutro_se_il_libro_non_e_flat(monkeypatch):
+    monkeypatch.setattr(telegram_daily, "book_state",
+                         lambda: (["  BTC: pos $300 → target $300"], False))
+    txt = telegram_daily.build()
+    assert "Stato trend" in txt and "Perche' non opera" not in txt
+
+
+def test_header_neutro_se_il_libro_non_e_leggibile(monkeypatch):
+    """Se book_state non sa dire se e' flat (eccezione -> None), niente header "non opera"
+    dichiarato senza saperlo davvero: si usa il neutro."""
+    monkeypatch.setattr(telegram_daily, "book_state",
+                         lambda: (["  ⚠️ stato libro non leggibile: Errore"], None))
+    txt = telegram_daily.build()
+    assert "Stato trend" in txt and "Perche' non opera" not in txt
 
 
 def test_nessuna_scrittura_ne_ordine():
@@ -52,3 +78,77 @@ def test_convenzione_trend_e_documentata():
     """La convenzione media-dei-segni (non frazione) e' la ragione per cui TP01 e' 0 con 1/3 su:
     se sparisce dal codice, il report mentirebbe sul 'quanto manca'."""
     assert "media dei SEGNI" in telegram_daily.trend_state.__doc__
+
+
+def test_soglia_accensione_2_negativi_1_positivo():
+    """2 negativi + 1 positivo: basta 1 flip -> soglia = il riferimento negativo PIU' BASSO."""
+    r = telegram_daily._soglia_accensione([-1, -1, 1], [100.0, 200.0])
+    assert r == 100.0
+
+
+def test_soglia_accensione_3_negativi():
+    """3 negativi: servono 2 flip -> soglia = il SECONDO riferimento negativo piu' basso, non il
+    minimo (il bug di #35 C3: superare solo il minimo lascia la media <= 0)."""
+    r = telegram_daily._soglia_accensione([-1, -1, -1], [100.0, 200.0, 300.0])
+    assert r == 200.0
+
+
+def test_soglia_accensione_1_negativo_2_positivi_gia_acceso():
+    """1 negativo + 2 positivi: la media e' gia' positiva (TP01 > 0) -> nessuna soglia da dare."""
+    assert telegram_daily._soglia_accensione([-1, 1, 1], [100.0]) is None
+
+
+def test_soglia_accensione_nessun_negativo():
+    """Tutti positivi: nessun riferimento negativo, nessuna soglia."""
+    assert telegram_daily._soglia_accensione([1, 1, 1], []) is None
+
+
+def test_soglia_accensione_servono_tutti_i_flip():
+    """2 negativi su 2 (nessun positivo): servono ENTRAMBI i flip -> k == len(neg_refs), soglia
+    = il piu' alto dei due (l'ultimo a bastare), non il minimo."""
+    r = telegram_daily._soglia_accensione([-1, -1], [200.0, 100.0])
+    assert r == 200.0
+
+
+# ---------------------------------------------------------------- (#35 C14) gate: data, formato
+
+
+def test_gate_date_legge_da_monitor_health():
+    """La data non e' una seconda dichiarazione: viene da monitor_health.MONITORS (`gate`),
+    la stessa tabella che gli alert dei sorveglianti citano come "METTE A RISCHIO il gate"."""
+    from src.live.monitor_health import MONITORS
+    m = next(x for x in MONITORS if x.name == "paper_statarb")
+    assert m.gate[:10] == "2026-09-27"
+    assert telegram_daily._gate_date("paper_statarb", date(2030, 1, 1)) == date(2026, 9, 27)
+    m2 = next(x for x in MONITORS if x.name == "paper_xsr")
+    assert telegram_daily._gate_date("paper_xsr", date(2030, 1, 1)) == date.fromisoformat(m2.gate[:10])
+
+
+def test_gate_date_usa_la_cablata_se_il_monitor_non_esiste():
+    """Rete di sicurezza: se il nome non e' in MONITORS (o cambia), non si inventa una data."""
+    assert telegram_daily._gate_date("non_esiste_piu'", date(2030, 1, 1)) == date(2030, 1, 1)
+
+
+def test_fmt_giorni_tre_casi_mai_il_doppio_segno():
+    """(#35 C14) giorni>0 -> '-Ng'; 0 -> 'oggi'; <0 -> 'scaduto da Ng'. Mai '(−-1g)'."""
+    assert telegram_daily._fmt_giorni(5) == "-5g"
+    assert telegram_daily._fmt_giorni(0) == "oggi"
+    assert telegram_daily._fmt_giorni(-1) == "scaduto da 1g"
+    assert "-1g" not in telegram_daily._fmt_giorni(-1).replace("scaduto da 1g", "")  # niente doppio segno
+
+
+def test_monitors_formatta_i_tre_casi_del_giorno(monkeypatch, tmp_path):
+    """I rami di monitors() (barra assente qui: file inesistenti in tmp_path) usano tutti
+    _fmt_giorni, non il vecchio '(−{giorni}g)' cablato: futuro, oggi e passato si distinguono."""
+    from datetime import timedelta
+    oggi = date.today()
+    monkeypatch.setattr(telegram_daily, "ROOT", tmp_path)
+    monkeypatch.setattr(telegram_daily, "GATES",
+                         [("FUTURO", oggi + timedelta(days=10), "data/nonexiste1/r.jsonl", "k"),
+                          ("OGGI", oggi, "data/nonexiste2/r.jsonl", "k"),
+                          ("PASSATO", oggi - timedelta(days=3), "data/nonexiste3/r.jsonl", "k")])
+    righe = telegram_daily.monitors()
+    assert "(-10g)" in righe[0]
+    assert "(oggi)" in righe[1]
+    assert "(scaduto da 3g)" in righe[2]
+    assert "−-" not in "".join(righe) and "(−" not in "".join(righe)   # mai il doppio segno

tests/test_trades_db_sync.py

@@ -0,0 +1,120 @@
+"""scripts/live/trades_db.py:sync() — controllo di conteggio jsonl-vs-DB (#37 C11, #35 A9).
+
+Dal 2026-08-24T01:07Z (`tradesdb.CUTOFF_TS_VERO`) `book_executions.jsonl` porta l'ora VERA del
+fill: un conteggio diverso da quello del DB in quella finestra segnala che una delle due fonti
+sta perdendo fill. `sync()` non ripara (come `reconcile()`): riporta e notifica UNA volta per
+episodio, con lo stesso marcatore-solo-su-invio-riuscito di fee_watch (#37 B9).
+"""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+sys.path.insert(0, str(ROOT / "scripts" / "live"))
+
+from src.live import tradesdb as T  # noqa: E402
+
+import trades_db as TD  # noqa: E402
+
+
+@pytest.fixture()
+def redirige(monkeypatch, tmp_path):
+    """DB, cron_book.log e jsonl finti in tmp_path: non si tocca mai il libro vero."""
+    reale = T.connect
+    monkeypatch.setattr(T, "connect", lambda *a, **k: reale(tmp_path / "t.db"))
+    monkeypatch.setattr(T, "CRON_LOG", tmp_path / "cron_book.log")       # assente -> nessun fill dal log
+    monkeypatch.setattr(T, "EXEC_JSONL", tmp_path / "book_executions.jsonl")
+    return tmp_path
+
+
+def _semina_db(tmp_path, n, ts=None):
+    ts = ts or T.CUTOFF_TS_VERO
+    con = T.connect(tmp_path / "t.db")
+    fills = [T.Fill(ts_utc=ts, asset="BTC", side="buy", qty=0.001 * (i + 1), price=70000.0, fee=0.01)
+             for i in range(n)]
+    T.upsert_fills(con, fills)
+    con.close()
+
+
+def _scrivi_jsonl(tmp_path, n, ts=None):
+    ts = ts or T.CUTOFF_TS_VERO
+    righe = [dict(ts_utc=ts, asset="BTC", side="buy", filled=0.001 * (i + 1), price=70000.0, fee=0.01)
+             for i in range(n)]
+    (tmp_path / "book_executions.jsonl").write_text(
+        "\n".join(json.dumps(r) for r in righe) + ("\n" if righe else ""))
+
+
+def test_conteggi_uguali_non_notifica(monkeypatch, redirige):
+    tmp_path = redirige
+    _semina_db(tmp_path, 2)
+    _scrivi_jsonl(tmp_path, 2)
+    inviato = []
+    monkeypatch.setattr(TD, "notify", lambda *a, **k: inviato.append((a, k)) or True)
+    r = TD.sync(verbose=False)
+    assert inviato == []
+    assert r["jsonl_dal_24_08"] == r["db_dal_24_08"] == 2
+
+
+def test_conteggi_diversi_notifica_una_volta_per_episodio(monkeypatch, redirige):
+    tmp_path = redirige
+    _semina_db(tmp_path, 2)
+    _scrivi_jsonl(tmp_path, 3)
+    inviato = []
+    monkeypatch.setattr(TD, "notify", lambda *a, **k: inviato.append((a, k)) or True)
+    r = TD.sync(verbose=False)
+    assert len(inviato) == 1, "conteggi diversi: un invio"
+    assert r["jsonl_dal_24_08"] == 3 and r["db_dal_24_08"] == 2
+
+    # secondo giro, stessa divergenza -> nessun invio aggiuntivo (marcatore gia' scritto)
+    r2 = TD.sync(verbose=False)
+    assert len(inviato) == 1, "stesso episodio: nessun secondo invio"
+    assert r2["jsonl_dal_24_08"] == 3 and r2["db_dal_24_08"] == 2
+
+
+def test_invio_fallito_non_marca_ritenta_al_giro_dopo(monkeypatch, redirige):
+    tmp_path = redirige
+    _semina_db(tmp_path, 2)
+    _scrivi_jsonl(tmp_path, 3)
+    inviato = []
+    monkeypatch.setattr(TD, "notify", lambda *a, **k: inviato.append((a, k)) or False)
+    TD.sync(verbose=False)
+    TD.sync(verbose=False)
+    assert len(inviato) == 2, "invio fallito -> marcatore non scritto, si ritenta ogni giro"
+
+
+def test_conteggio_rientrato_azzera_il_marcatore(monkeypatch, redirige):
+    tmp_path = redirige
+    _semina_db(tmp_path, 2)
+    _scrivi_jsonl(tmp_path, 3)
+    monkeypatch.setattr(TD, "notify", lambda *a, **k: True)
+    TD.sync(verbose=False)                 # divergenza detta
+
+    # il jsonl si allinea (es. arrivato il fill mancante) -> episodio chiuso
+    _scrivi_jsonl(tmp_path, 2)
+    inviato = []
+    monkeypatch.setattr(TD, "notify", lambda *a, **k: inviato.append((a, k)) or True)
+    TD.sync(verbose=False)
+    assert inviato == [], "conteggi tornati uguali: nessun invio da fare"
+
+    # una NUOVA divergenza deve poter notificare di nuovo (il marcatore era stato azzerato)
+    _scrivi_jsonl(tmp_path, 4)
+    TD.sync(verbose=False)
+    assert len(inviato) == 1
+
+
+def test_fill_prima_del_24_08_non_entra_nel_conteggio(monkeypatch, redirige):
+    """Prima di CUTOFF_TS_VERO il ts_utc del jsonl e' la barra di segnale, non l'ora vera: non
+    e' comparabile col DB, e non deve contare come divergenza."""
+    tmp_path = redirige
+    _semina_db(tmp_path, 0)
+    _scrivi_jsonl(tmp_path, 2, ts="2026-08-10T00:00:00+00:00")   # prima della soglia
+    inviato = []
+    monkeypatch.setattr(TD, "notify", lambda *a, **k: inviato.append((a, k)) or True)
+    r = TD.sync(verbose=False)
+    assert inviato == []
+    assert r["jsonl_dal_24_08"] == 0 and r["db_dal_24_08"] == 0

tests/test_venue_probe.py

@@ -8,17 +8,17 @@ Cosa difendono, in ordine di gravita':
      da `place-failed`. Sono due guasti diversi (P5) e solo uno lascia la posizione scoperta.
   2. **isolamento per asset**: un guasto su BTC non deve impedire ad ETH di essere protetto.
   3. **la diagnosi**: manutenzione, gateway e "non vedo" devono essere distinguibili (P4/P5).
-  4. **P9**: `atteso` declassa la gravita' SOLO su evidenza di manutenzione, e SOLO dentro lo slot.
+  4. **`atteso` e' sempre False** (#35 B11): il ramo che lo declassava (finestra di release del
+     martedi') e' stato rimosso perche' l'unico chiamante (cron :47) non ci cadeva mai dentro.
 """
 import sys
-from datetime import datetime, timezone
 from pathlib import Path
 
 PROJECT_ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(PROJECT_ROOT))
 
 from src.live.venue_probe import (V_GATEWAY, V_IGNOTO, V_MANUTENZIONE, V_VENUE_GIU, diagnose,
-                                  errori_dal_report, in_release_window)
+                                  errori_dal_report)
 
 MANUTENZIONE_GREZZA = ('HTTP 503 {"jsonrpc":"2.0","error":'
                        '{"message":"system_maintenance","code":11051}}')
@@ -94,39 +94,19 @@ def test_la_prova_viaggia_sempre_con_il_verdetto():
 
 
 # ---------------------------------------------------------------------------
-# 2. P9 — `atteso` declassa, ma solo su EVIDENZA e solo DENTRO lo slot.
+# 2. `atteso` e' sempre False (#35 B11) — nessun verdetto, manutenzione compresa, si declassa.
 # ---------------------------------------------------------------------------
-def test_finestra_di_release_martedi_nove_utc():
-    dentro = datetime(2026, 8, 25, 9, 7, tzinfo=timezone.utc)      # martedi', l'episodio vero
-    assert in_release_window(dentro) is True
-    assert in_release_window(datetime(2026, 8, 25, 9, 0, tzinfo=timezone.utc)) is True
-    assert in_release_window(datetime(2026, 8, 25, 9, 29, tzinfo=timezone.utc)) is True
-    assert in_release_window(datetime(2026, 8, 25, 9, 30, tzinfo=timezone.utc)) is False   # sforata
-    assert in_release_window(datetime(2026, 8, 25, 8, 59, tzinfo=timezone.utc)) is False
-    assert in_release_window(datetime(2026, 8, 26, 9, 7, tzinfo=timezone.utc)) is False    # mercoledi'
-
-
-def test_manutenzione_dentro_lo_slot_e_attesa_e_declassa():
-    d = diagnose([], now=datetime(2026, 8, 25, 9, 7, tzinfo=timezone.utc),
-                 sonda=_sonda_manutenzione)
-    assert d.verdetto == V_MANUTENZIONE and d.atteso is True
-    assert d.gravita == "ℹ️"                        # declassato, NON silenziato
-    assert "ATTESO" in d.riga()
-
-
-def test_manutenzione_che_sfora_lo_slot_torna_rumorosa():
-    """Il 18/08 Deribit annuncio' 15-30 minuti e resto' giu' oltre l'ora: alle 10:07 la
-    manutenzione non era piu' l'evento annunciato, ed e' di nuovo una notizia."""
-    d = diagnose([], now=datetime(2026, 8, 18, 10, 7, tzinfo=timezone.utc),
-                 sonda=_sonda_manutenzione)
+def test_manutenzione_non_si_declassa_piu_atteso_sempre_falso():
+    """Il ramo che declassava (finestra di release del martedi') e' rimosso: una manutenzione
+    riconosciuta dalla sonda resta comunque `atteso is False` e alla gravita' piena."""
+    d = diagnose([], sonda=_sonda_manutenzione)
     assert d.verdetto == V_MANUTENZIONE and d.atteso is False
-    assert d.gravita != "ℹ️"
+    assert d.gravita == "⚠️"
+    assert "ATTESO" not in d.riga()
 
 
-def test_un_guasto_vero_di_martedi_mattina_non_viene_declassato():
-    """Il rischio della finestra: costruire il silenzio proprio nell'ora in cui serve. Un gateway
-    rotto dentro lo slot di release deve restare al massimo della gravita'."""
-    d = diagnose([], now=datetime(2026, 8, 25, 9, 7, tzinfo=timezone.utc), sonda=_sonda_ok)
+def test_un_guasto_vero_non_e_mai_atteso():
+    d = diagnose(["BTC: fallback close (HTTPError)"], sonda=_sonda_ok)
     assert d.verdetto == V_GATEWAY and d.atteso is False and d.gravita == "🛑"