Adriano Dal Pastro
418f523be7
Decisione dell'operatore: GTAA01 NON si blocca oggi, si decide quando il book arriva a $15k. Il motivo registrato non e' «non contribuisce» — la misura dice il contrario (+0,095/+0,124 di Sharpe a iso-rischio secondo la fase, positivo in tutte e cinque, hold-out +0,247, 6 anni su 8, correlazione col resto +0,087). Il motivo e' che sotto $15k di book lo sleeve NON E' ACCENDIBILE: GTAA_MIN_CAPITAL e' $3.000 allocati, che al peso 20% fa $15.000 contro i $2.068 attuali. Finora era manutenzione senza beneficio incassabile. Registrate anche le due ragioni contro il blocco, che restano valide: N7 (uno sleeve difensivo si giudica sul sinistro, e questa finestra non lo contiene) e N4 (togliendolo il portafoglio di ricerca diventa 100% cripto su un venue solo). E il costo che sarebbe andato pagato: GTAA01 e' uno dei quattro nomi di W_DEPLOY, da cui escono i $313k del muro. PERCHE' LA DECISIONE NON SI DIMENTICHI (N9). Una decisione parcheggiata su un numero che nessuno sorveglia e' parcheggiata per sempre. `journal.SOGLIE_CAPITALE` legge l'equity ogni giorno — il giornale lo fa comunque — e il giorno che supera la soglia la voce dice quale decisione si sblocca e perche'. Seconda riga: $20k, dove si riapre «100% Deribit fino a $20k». Una riga si TOGLIE quando la decisione e' presa. 5 test, incluso «equity non leggibile non e' una soglia superata» (P5). GATE (A) RITIRATO come pass/fail, congelando il MOTIVO e non l'esito — come il 07/08 col confronto fra ranghi, e per la stessa ragione: il criterio decide su un margine piu' piccolo del rumore che lo scuote. In piu' una guardia sulla CAUSA (`test_la_fase_di_ribilanciamento_e_ancora_ancorata_alla_POSIZIONE`) che si rompe il giorno che qualcuno ancorasse la fase al calendario: quel giorno il gate potrebbe tornare decidibile, ed e' un fatto da guardare, non da ignorare. Il test sul margine assoluto si e' auto-ritirato: diceva «se scendesse sotto un centesimo questo criterio smetterebbe di essere una misura», ed e' sceso a 0,0074. CLAUDE.md: §3 nuova riga · §5.2 RISOLTO (trasporto allarmi) · §5.4 riscritto (la domanda fiscale (a) ha una risposta alla fonte: derivati in c-quater al 26%, non c-sexies al 33% — Circolare AdE 30/E del 27/10/2023, citata verbatim) · §5.13 nuovo debito, la fase che ruota — quantificata e dichiarata INNOCUA oggi (D5): 0,029 di Sharpe a livello di portafoglio, dentro la banda [1,81-2,12]. Diario: docs/diary/2026-08-28-gtaa-fase-e-allarmi.md Suite: 789 passati, 0 falliti. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
504 lines
24 KiB
Python
504 lines
24 KiB
Python
"""Il giornale di bordo: tre stati invece di due, e la nota dell'operatore che non si perde.
|
|
|
|
Il rischio di un giornale automatico non e' sbagliare un numero: e' scrivere `0` dove il dato
|
|
non c'era, e leggerlo fra sei mesi come una misura.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.live import journal as J # noqa: E402
|
|
from src.live import tradesdb as T # noqa: E402
|
|
|
|
|
|
@pytest.fixture()
|
|
def db(tmp_path):
|
|
con = T.connect(tmp_path / "t.db")
|
|
yield con
|
|
con.close()
|
|
|
|
|
|
# ------------------------------------------------- i giri attesi non sono sempre 24
|
|
|
|
def test_giorno_in_corso_non_e_un_giorno_bucato(db, monkeypatch):
|
|
"""24 giri attesi su un giorno non finito = allarme a ogni esecuzione, cioe' rumore."""
|
|
oggi = datetime.now(timezone.utc).date()
|
|
monkeypatch.setattr(J, "stato_libro", lambda c, g: dict(giri=datetime.now(timezone.utc).hour + 1,
|
|
feed_skh_min=0))
|
|
s = J.salute(db, oggi)
|
|
assert s["giorno_in_corso"] is True
|
|
assert s["atteso_giri"] == datetime.now(timezone.utc).hour + 1
|
|
assert s["giri_mancanti"] == 0
|
|
|
|
|
|
def test_giorno_chiuso_si_giudica_su_24(db, monkeypatch):
|
|
monkeypatch.setattr(J, "stato_libro", lambda c, g: dict(giri=20, feed_skh_min=0))
|
|
s = J.salute(db, date(2026, 8, 21))
|
|
assert s["giorno_in_corso"] is False
|
|
assert s["atteso_giri"] == 24 and s["giri_mancanti"] == 4
|
|
|
|
|
|
def test_controllo_positivo_un_giorno_davvero_bucato_viene_segnalato(db, monkeypatch):
|
|
monkeypatch.setattr(J, "stato_libro", lambda c, g: dict(giri=6, feed_skh_min=None))
|
|
assert J.salute(db, date(2026, 8, 21))["giri_mancanti"] == 18
|
|
|
|
|
|
# ------------------------------------------------- dato assente != zero
|
|
|
|
def test_dvol_assente_e_None_con_ragione_mai_zero(db, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(J, "PROJECT_ROOT", tmp_path) # nessun file DVOL qui
|
|
(tmp_path / "data" / "raw").mkdir(parents=True)
|
|
m = J.metriche_mercato(date(2026, 8, 21))
|
|
for a in ("BTC", "ETH"):
|
|
if "errore" in m[a]:
|
|
continue
|
|
assert m[a]["dvol"] is None
|
|
assert m[a]["dvol_nota"] == "file DVOL assente" # la ragione, non un valore
|
|
|
|
|
|
def test_giorno_senza_giri_e_un_errore_dichiarato_non_un_libro_vuoto(db, monkeypatch):
|
|
monkeypatch.setattr(T, "CRON_LOG", Path("/tmp/non-esiste-affatto.log"))
|
|
lb = J.stato_libro(db, date(2026, 8, 21))
|
|
assert "errore" in lb and lb["giri"] == 0
|
|
assert "equity" not in lb # non inventa uno zero
|
|
|
|
|
|
def test_pnl_senza_letture_di_equity_non_fabbrica_un_delta(db):
|
|
p = J.pnl_giorno(db, date(2026, 8, 21))
|
|
assert p["delta_equity"] is None and p["equity_fine"] is None
|
|
assert p["realizzato_netto"] == 0.0 and p["roundtrip_chiusi"] == 0 # 0 trade e' un fatto
|
|
|
|
|
|
# ------------------------------------------------- la nota dell'operatore
|
|
|
|
def test_la_nota_sopravvive_a_un_ricalcolo(db):
|
|
g = date(2026, 8, 21)
|
|
voce = dict(giorno=g.isoformat(), ts_scritto=T.ora(), mercato={}, libro={}, pnl={}, salute={})
|
|
J.salva(db, voce, nota="rally post-CPI, non ho toccato niente")
|
|
J.salva(db, voce) # ricalcolo automatico
|
|
n = db.execute("SELECT nota FROM journal WHERE giorno=?", (g.isoformat(),)).fetchone()["nota"]
|
|
assert n == "rally post-CPI, non ho toccato niente"
|
|
|
|
|
|
def test_la_nota_si_puo_sostituire_esplicitamente(db):
|
|
g = date(2026, 8, 21)
|
|
voce = dict(giorno=g.isoformat(), ts_scritto=T.ora(), mercato={}, libro={}, pnl={}, salute={})
|
|
J.salva(db, voce, nota="prima")
|
|
J.salva(db, voce, nota="seconda")
|
|
n = db.execute("SELECT nota FROM journal WHERE giorno=?", (g.isoformat(),)).fetchone()["nota"]
|
|
assert n == "seconda"
|
|
|
|
|
|
# ------------------------------------------------- il markdown non mente
|
|
|
|
def test_il_markdown_dice_n_d_e_non_zero_quando_manca_il_dato(db):
|
|
voce = dict(giorno="2026-08-21", ts_scritto=T.ora(),
|
|
mercato={"BTC": {"errore": "feed non leggibile: FileNotFoundError"},
|
|
"ETH": {"errore": "feed non leggibile: FileNotFoundError"}},
|
|
libro={"errore": "nessun giro di book_execute quel giorno", "giri": 0},
|
|
pnl=J.pnl_giorno(db, date(2026, 8, 21)),
|
|
salute=dict(giri_book=0, feed_skh_min=None, atteso_giri=24,
|
|
giorno_in_corso=False, giri_mancanti=24))
|
|
J.salva(db, voce)
|
|
md = J.rendi_markdown(db, voce)
|
|
assert "n/d" in md and "24 mancanti" in md
|
|
assert "non misurata" in md # eta' feed sconosciuta
|
|
assert "$+0.00 di equity" not in md # niente zeri inventati
|
|
|
|
|
|
def test_il_markdown_riporta_il_segnale_che_il_libro_leggeva(db):
|
|
voce = dict(giorno="2026-08-21", ts_scritto=T.ora(), mercato={},
|
|
libro=dict(giri=24, ultimo_giro="2026-08-21T23:07:01+00:00", equity=644.0,
|
|
barra="2026-08-21", feed_skh_min=0, nozionale_lordo=196.0,
|
|
leva_lorda=0.304,
|
|
asset={"BTC": dict(tp_frac=0.176, skh_sign=1, skh_entry=75280.0,
|
|
target=123.0, posizione=126.0, azione="HOLD (a target)")}),
|
|
pnl=J.pnl_giorno(db, date(2026, 8, 21)),
|
|
salute=dict(giri_book=24, feed_skh_min=0, atteso_giri=24,
|
|
giorno_in_corso=False, giri_mancanti=0))
|
|
J.salva(db, voce)
|
|
md = J.rendi_markdown(db, voce)
|
|
assert "+0.176" in md and "LONG @ 75,280.0" in md and "0.30x" in md
|
|
|
|
|
|
def test_lo_stato_tsmom_usa_gli_orizzonti_di_produzione():
|
|
"""Il giornale deve riportare CIO' CHE IL LIBRO LEGGE, non un trend a caso."""
|
|
from src.strategies import trend_portfolio as tp
|
|
src = Path(tp.__file__).read_text()
|
|
for n in J.ORIZZONTI:
|
|
assert str(n) in src, f"orizzonte {n} non compare in trend_portfolio.py"
|
|
|
|
|
|
# =============================================================================================
|
|
# LETTURA RAGIONATA — ogni regola ha un test che la ACCENDE e uno che la tiene SPENTA
|
|
# =============================================================================================
|
|
# Una regola che si accende sempre non sta leggendo niente, e una che non si accende mai e'
|
|
# indistinguibile da una rotta. Servono entrambe le direzioni.
|
|
|
|
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 _ids(righe):
|
|
return [r["id"] for r in righe]
|
|
|
|
|
|
def test_libro_flat_dice_anche_PERCHE(db):
|
|
v = _voce(libro=_libro(BTC={}, ETH={}), pnl=_pnl())
|
|
r = J.lettura(db, v)
|
|
assert "stato" in _ids(r) and "perche_flat" in _ids(r)
|
|
testo = next(x["testo"] for x in r if x["id"] == "perche_flat")
|
|
assert "TP01 a zero" in testo and "SKH01 senza breakout" in testo
|
|
|
|
|
|
def test_libro_a_mercato_non_spiega_un_flat_che_non_c_e(db):
|
|
v = _voce(libro=_libro(BTC=dict(tp_frac=0.15, skh_sign=1, posizione=116.0)), pnl=_pnl())
|
|
r = J.lettura(db, v)
|
|
assert "perche_flat" not in _ids(r)
|
|
assert "$+116" in next(x["testo"] for x in r if x["id"] == "stato")
|
|
|
|
|
|
def test_disaccordo_si_accende_a_2_su_3_e_tace_a_3_su_3(db):
|
|
lb = _libro(BTC=dict(tp_frac=0.15, skh_sign=1, posizione=116.0))
|
|
misto = {"BTC": dict(chiusura=77000.0, tsmom_su=(2, 3), tsmom={})}
|
|
pieno = {"BTC": dict(chiusura=77000.0, tsmom_su=(3, 3), tsmom={})}
|
|
assert "disaccordo" in _ids(J.lettura(db, _voce(mercato=misto, libro=lb, pnl=_pnl())))
|
|
assert "disaccordo" not in _ids(J.lettura(db, _voce(mercato=pieno, libro=lb, pnl=_pnl())))
|
|
|
|
|
|
def test_incoerenza_trend_su_ma_libro_fuori(db):
|
|
lb = _libro(BTC=dict(tp_frac=0.0, skh_sign=0, posizione=0.0))
|
|
m = {"BTC": dict(chiusura=77000.0, tsmom_su=(3, 3), tsmom={})}
|
|
r = J.lettura(db, _voce(mercato=m, libro=lb, pnl=_pnl()))
|
|
riga = next(x for x in r if x["id"] == "incoerenza")
|
|
assert riga["livello"] == "attenzione"
|
|
|
|
|
|
def test_il_tetto_di_leva_e_DERIVATO_dal_config_non_riscritto(db, monkeypatch, tmp_path):
|
|
"""Se qualcuno cambia `max_notional_per_asset_frac`, la lettura deve cambiare da sola."""
|
|
(tmp_path / "config").mkdir()
|
|
(tmp_path / "config" / "live.json").write_text(json.dumps({"max_notional_per_asset_frac": 0.25}))
|
|
monkeypatch.setattr(J, "PROJECT_ROOT", tmp_path)
|
|
lb = _libro(BTC=dict(posizione=150.0), ETH=dict(posizione=150.0))
|
|
r = J.lettura(db, _voce(libro=lb, pnl=_pnl()))
|
|
testo = next(x["testo"] for x in r if x["id"] == "leva")
|
|
assert "0.50x" in testo and "0.25" in testo # 0.25 x 2 asset, NON 1.00x
|
|
|
|
|
|
def test_config_illeggibile_lo_dice_invece_di_inventare_un_tetto(db, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(J, "PROJECT_ROOT", tmp_path) # nessun config qui
|
|
r = J.lettura(db, _voce(libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl()))
|
|
riga = next(x for x in r if x["id"] == "leva")
|
|
assert "non verificabile" in riga["testo"] and riga["livello"] == "attenzione"
|
|
|
|
|
|
def test_soglia_IV_rank_derivata_dal_sleeve(db):
|
|
"""La soglia del gate viene da VRP_CFG, e l'IV-rank e' quello ESPANDENTE del sleeve."""
|
|
from src.portfolio.sleeves import VRP_CFG
|
|
lb = _libro(BTC=dict(posizione=100.0))
|
|
m = {"BTC": dict(chiusura=77000.0, dvol=42.0, rv30_annua=43.0, dvol_rank_exp=0.18)}
|
|
testo = next(x["testo"] for x in J.lettura(db, _voce(mercato=m, libro=lb, pnl=_pnl()))
|
|
if x["id"] == "vol")
|
|
assert f"{VRP_CFG['gate_ivr']:.2f}" in testo and "sotto la soglia" in testo
|
|
m["BTC"]["dvol_rank_exp"] = 0.90
|
|
testo = next(x["testo"] for x in J.lettura(db, _voce(mercato=m, libro=lb, pnl=_pnl()))
|
|
if x["id"] == "vol")
|
|
assert "**sopra**" in testo
|
|
|
|
|
|
def test_iv_rank_espandente_e_diverso_dal_rank_a_un_anno():
|
|
"""I due rank rispondono a domande diverse: il 2026-08-23 danno 0.18 contro 0.52.
|
|
|
|
Etichettare l'uno col nome dell'altro ribalta il verdetto sul gate di VRP01.
|
|
"""
|
|
m = J.metriche_mercato(date.today())
|
|
for a in ("BTC", "ETH"):
|
|
d = m.get(a) or {}
|
|
if d.get("dvol_rank_exp") is None or d.get("dvol_rank_1a") is None:
|
|
continue
|
|
assert d["dvol_rank_exp"] != d["dvol_rank_1a"]
|
|
|
|
|
|
def test_regola_fee_quando_il_costo_supera_il_movimento(db):
|
|
lb = _libro(BTC=dict(posizione=100.0))
|
|
r = J.lettura(db, _voce(libro=lb, pnl=_pnl(delta_equity=0.01, fill=2, fee=0.05)))
|
|
assert "fee" in _ids(r)
|
|
r = J.lettura(db, _voce(libro=lb, pnl=_pnl(delta_equity=5.00, fill=2, fee=0.05)))
|
|
assert "fee" not in _ids(r) and "pnl" in _ids(r)
|
|
|
|
|
|
def test_evento_si_accende_oltre_2_deviazioni_e_tace_sotto(db):
|
|
lb = _libro(BTC=dict(posizione=100.0))
|
|
# RV30 36.5% annua -> sd giornaliera ~1.91%
|
|
grosso = {"BTC": dict(chiusura=77000.0, ret_1g=-6.0, rv30_annua=36.5)}
|
|
normale = {"BTC": dict(chiusura=77000.0, ret_1g=-1.0, rv30_annua=36.5)}
|
|
assert "evento" in _ids(J.lettura(db, _voce(mercato=grosso, libro=lb, pnl=_pnl())))
|
|
assert "evento" not in _ids(J.lettura(db, _voce(mercato=normale, libro=lb, pnl=_pnl())))
|
|
|
|
|
|
def test_le_regole_tacciono_sul_non_misurato(db):
|
|
"""Mercato assente: nessuna riga su vol o eventi. Il silenzio e' il comportamento giusto."""
|
|
r = J.lettura(db, _voce(mercato={"BTC": {"errore": "feed KO"}},
|
|
libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl(delta_equity=None)))
|
|
assert "vol" not in _ids(r) and "evento" not in _ids(r) and "pnl" not in _ids(r)
|
|
|
|
|
|
def test_giri_mancanti_e_feed_vecchio_sono_attenzioni(db):
|
|
s = dict(giri_book=6, feed_skh_min=685, atteso_giri=24, giorno_in_corso=False, giri_mancanti=18)
|
|
r = J.lettura(db, _voce(libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl(), salute=s))
|
|
liv = {x["id"]: x["livello"] for x in r}
|
|
assert liv.get("giri") == "attenzione" and liv.get("feed") == "attenzione"
|
|
|
|
|
|
def test_senza_giri_la_lettura_dice_NON_MISURATO_e_si_ferma(db):
|
|
r = J.lettura(db, _voce())
|
|
assert _ids(r) == ["stato"] and "non e' stato misurato" in r[0]["testo"]
|
|
|
|
|
|
def test_il_bucket_di_convinzione_non_inventa_i_due_terzi():
|
|
"""`tsmom_blend` media tre np.sign(): dopo il clip long-flat la direzione sta in {0,1/3,1}."""
|
|
assert J._bucket_convinzione(0.0) == "flat"
|
|
assert J._bucket_convinzione(0.15) == "parziale"
|
|
assert J._bucket_convinzione(0.90) == "piena"
|
|
|
|
|
|
def test_concentrazione_tace_su_un_cumulato_irrilevante(db):
|
|
"""Vera ma inutile: su pochi dollari la concentrazione e' aritmetica, non informazione."""
|
|
for i in range(60):
|
|
db.execute("INSERT INTO equity VALUES (?,?,?)",
|
|
(f"2026-06-{(i % 28) + 1:02d}T{i % 24:02d}:00:00+00:00", 600.0 + i * 0.03, "t"))
|
|
db.commit()
|
|
v = _voce(libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl(), giorno="2026-06-28")
|
|
assert "concentrazione" not in _ids(J.lettura(db, v)) # ~$1.8 di cumulato
|
|
|
|
|
|
def test_concentrazione_si_accende_quando_la_taglia_conta(db):
|
|
for i in range(60):
|
|
v = 600.0 if i < 50 else 600.0 + (i - 49) * 4.0 # salto nella coda
|
|
db.execute("INSERT INTO equity VALUES (?,?,?)",
|
|
(f"2026-06-{(i % 28) + 1:02d}T{i % 24:02d}:00:00+00:00", v, "t"))
|
|
db.commit()
|
|
v = _voce(libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl(), giorno="2026-06-28")
|
|
assert "concentrazione" in _ids(J.lettura(db, v))
|
|
|
|
|
|
# ------------------------------------------- il giorno IN CORSO non si presenta come chiuso
|
|
#
|
|
# Una pagina di giorno in corso ha TUTTE le sezioni di una chiusa: passa ogni controllo di
|
|
# completezza e freschezza, e si legge come una giornata. Fino al 2026-08-25 lo diceva solo
|
|
# la §Salute — quinta sezione su otto — mentre titolo, P&L e Lettura parlavano al passato di
|
|
# una frazione di giornata. Le due prove sono accoppiate: una accende la marcatura, l'altra
|
|
# verifica che su un giorno chiuso resti SPENTA (una marcatura sempre accesa non si legge).
|
|
|
|
def _voce_salute(con, in_corso: bool, giri: int):
|
|
return dict(giorno="2026-08-25", ts_scritto=T.ora(), mercato={},
|
|
libro=dict(giri=giri, ultimo_giro="2026-08-25T07:07:01+00:00", equity=642.0,
|
|
barra="2026-08-25", feed_skh_min=0, nozionale_lordo=200.0,
|
|
leva_lorda=0.31, asset={}),
|
|
pnl=J.pnl_giorno(con, date(2026, 8, 25)),
|
|
salute=dict(giri_book=giri, feed_skh_min=0, atteso_giri=giri if in_corso else 24,
|
|
giorno_in_corso=in_corso, giri_mancanti=0))
|
|
|
|
|
|
def test_una_pagina_di_giorno_in_corso_si_dichiara_parziale_nel_titolo(db):
|
|
v = _voce_salute(db, in_corso=True, giri=8)
|
|
J.salva(db, v)
|
|
md = J.rendi_markdown(db, v)
|
|
titolo = md.splitlines()[0]
|
|
assert "PARZIALE" in titolo and "giorno in corso" in titolo # nel posto che si legge
|
|
testata = "\n".join(md.splitlines()[:6])
|
|
assert "NON e' chiusa" in testata # e non solo nel titolo
|
|
assert "8 giri su 24" in testata # con la copertura dichiarata
|
|
assert "non si aggiorna da sola" in testata # e col fatto che e' congelata
|
|
|
|
|
|
def test_una_pagina_di_giorno_chiuso_non_si_dichiara_parziale(db):
|
|
v = _voce_salute(db, in_corso=False, giri=24)
|
|
J.salva(db, v)
|
|
md = J.rendi_markdown(db, v)
|
|
assert "PARZIALE" not in md and "non si aggiorna da sola" not in md
|
|
assert md.splitlines()[0] == "# Giornale di bordo — 2026-08-25"
|
|
|
|
|
|
def test_il_cron_non_apre_piu_la_voce_del_giorno_in_corso():
|
|
"""Guardia sul SORGENTE: `journal.py` senza `--giorno` scrive OGGI, e alle 00:30 UTC oggi
|
|
e' mezz'ora. Chi la rimettesse congelerebbe di nuovo una pagina che sembra una giornata.
|
|
Il test non vieta di rigenerarla spesso: vieta di scriverla UNA volta e lasciarla li'.
|
|
"""
|
|
cron = (ROOT / "scripts" / "cron_daily.sh").read_text()
|
|
chiamate = [r.strip() for r in cron.splitlines()
|
|
if "scripts/live/journal.py" in r and not r.strip().startswith("#")]
|
|
assert chiamate, "il giornale non gira piu' nel cron: e' un'altra rottura, non questa"
|
|
for r in chiamate:
|
|
assert "--giorno" in r, f"apre un giorno implicito (= oggi, parziale): {r}"
|
|
|
|
|
|
# ------------------------------------------------- movimenti di capitale nel P&L
|
|
# Il difetto (2026-08-25): la voce dichiarava «giorno: +$1.414,57» quando $1.399 erano un
|
|
# deposito USDC — il P&L era un delta di equity e ogni versamento ci finiva dentro come
|
|
# profitto, per sempre (il cumulato dall'arming non si sarebbe mai piu' corretto).
|
|
|
|
def _px_piatto(livello=100.0):
|
|
ix = pd.date_range("2026-08-20", "2026-08-27", freq="1h", tz="UTC")
|
|
return pd.Series(livello, index=ix)
|
|
|
|
|
|
def _serie_equity(con, punti):
|
|
for ts, eq in punti:
|
|
con.execute("INSERT INTO equity (ts_utc, equity, src) VALUES (?,?,?)", (ts, eq, "test"))
|
|
con.commit()
|
|
|
|
|
|
def test_un_versamento_viene_rilevato_e_scorporato(db, monkeypatch):
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-25T09:47:01+00:00", 667.0),
|
|
("2026-08-25T10:47:01+00:00", 667.5),
|
|
("2026-08-25T11:47:01+00:00", 2066.5),
|
|
("2026-08-25T12:47:01+00:00", 2064.0)])
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
ev = p["movimenti_giorno"]
|
|
assert len(ev) == 1 and ev[0]["classe"] == "movimento"
|
|
assert p["mov_giorno"] == pytest.approx(1399.0)
|
|
# il delta resta il FATTO, il trading e' il netto
|
|
assert p["delta_equity"] == pytest.approx(2064.0 - 667.0)
|
|
assert p["trading_giorno"] == pytest.approx(p["delta_equity"] - 1399.0)
|
|
assert p["trading_da_arming"] == pytest.approx(p["cumulato_da_arming"] - 1399.0)
|
|
|
|
|
|
def test_il_markdown_scorpora_e_lo_dice(db, monkeypatch):
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 667.0),
|
|
("2026-08-25T11:47:01+00:00", 2066.0)])
|
|
voce = _voce(pnl=J.pnl_giorno(db, date(2026, 8, 25)), giorno="2026-08-25")
|
|
md = J.rendi_markdown(db, voce)
|
|
assert "movimenti di capitale" in md and "trading" in md
|
|
righe = J.lettura(db, voce)
|
|
assert "movimento" in _ids(righe)
|
|
|
|
|
|
def test_un_crash_a_tutta_leva_resta_AMBIGUO_e_non_diventa_prelievo(db, monkeypatch):
|
|
# controllo positivo al contrario (M15): il mercato fa -15% nell'ora, il libro al tetto
|
|
# di leva (1.0x) puo' spiegare il -14% di equity -> NON e' un prelievo, e' dichiarato.
|
|
ix = pd.date_range("2026-08-20", "2026-08-27", freq="1h", tz="UTC")
|
|
px = pd.Series(100.0, index=ix)
|
|
px[ix >= "2026-08-25T11:00:00+00:00"] = 85.0
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: px)
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 600.0),
|
|
("2026-08-25T11:47:01+00:00", 516.0)])
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
assert p["movimenti_giorno"][0]["classe"] == "ambiguo"
|
|
assert p["mov_giorno"] == 0.0 # niente scorporo
|
|
assert p["trading_giorno"] == pytest.approx(p["delta_equity"])
|
|
righe = J.lettura(db, _voce(pnl=p, giorno="2026-08-25"))
|
|
amb = [r for r in righe if r["id"] == "movimento_ambiguo"]
|
|
assert amb and amb[0]["livello"] == "attenzione"
|
|
|
|
|
|
def test_sotto_soglia_niente_eventi_e_trading_uguale_al_delta(db, monkeypatch):
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 600.0),
|
|
("2026-08-25T11:47:01+00:00", 612.0)]) # +2%: sotto EQUITY_JUMP_ALERT
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
assert p["movimenti_giorno"] == [] and p["mov_giorno"] == 0.0
|
|
assert p["trading_giorno"] == pytest.approx(p["delta_equity"])
|
|
md = J.rendi_markdown(db, _voce(pnl=p, giorno="2026-08-25"))
|
|
assert "movimenti di capitale" not in md
|
|
|
|
|
|
def test_la_soglia_e_DERIVATA_dal_book_non_riscritta(db, monkeypatch):
|
|
# P1: se il rilevatore del book cambia soglia, il giornale la segue senza essere toccato.
|
|
from src.live import book as B
|
|
monkeypatch.setattr(B, "EQUITY_JUMP_ALERT", 0.50)
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 600.0),
|
|
("2026-08-25T11:47:01+00:00", 780.0)]) # +30%: sotto la soglia alzata
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
assert p["movimenti_giorno"] == []
|
|
|
|
|
|
def test_mercato_non_misurabile_e_ambiguo_dichiarato_non_movimento(db, monkeypatch):
|
|
# P5: "non vedo" non e' "va tutto bene" — senza feed il salto NON si attribuisce.
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: None)
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 667.0),
|
|
("2026-08-25T11:47:01+00:00", 2066.0)])
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
e = p["movimenti_giorno"][0]
|
|
assert e["classe"] == "ambiguo" and e["mercato_max_pct"] is None
|
|
righe = J.lettura(db, _voce(pnl=p, giorno="2026-08-25"))
|
|
assert any(r["id"] == "movimento_ambiguo" and "non e' misurabile" in r["testo"] for r in righe)
|
|
|
|
|
|
def test_la_concentrazione_non_conta_i_versamenti_come_pnl(db, monkeypatch):
|
|
# 30 giorni piatti a 600, poi un versamento da 1400: senza scorporo la regola direbbe
|
|
# "tutto il P&L viene dagli ultimi 7 giorni" su un P&L che non esiste.
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
punti = [(f"2026-07-{g:02d}T12:00:00+00:00", 600.0) for g in range(1, 32)]
|
|
punti += [(f"2026-08-{g:02d}T12:00:00+00:00", 600.0) for g in range(1, 25)]
|
|
punti += [("2026-08-25T11:47:01+00:00", 2000.0)]
|
|
_serie_equity(db, punti)
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
righe = J.lettura(db, _voce(pnl=p, giorno="2026-08-25"))
|
|
conc = [r for r in righe if r["id"] == "concentrazione"]
|
|
assert conc == [] # il cumulato di TRADING e' 0: la regola deve tacere
|
|
|
|
|
|
# =============================================================================================
|
|
# DECISIONI PARCHEGGIATE SU UNA SOGLIA DI CAPITALE
|
|
# =============================================================================================
|
|
# Il 2026-08-28 la decisione «GTAA01: tenere o bloccare» e' stata rinviata a $15k di book, che e'
|
|
# la soglia sotto la quale lo sleeve non e' deployabile. Una decisione parcheggiata su un numero
|
|
# che nessuno sorveglia e' parcheggiata per sempre (N9): questi test blindano la sorveglianza.
|
|
|
|
def test_sotto_soglia_non_dice_niente():
|
|
assert J.soglie_superate(2_068.0) == []
|
|
|
|
|
|
def test_la_soglia_gtaa_e_15k_e_scatta_quando_l_equity_arriva():
|
|
soglie = [s for s, _ in J.SOGLIE_CAPITALE]
|
|
assert 15_000.0 in soglie, "la soglia di GTAA01 non e' piu' dichiarata"
|
|
assert J.soglie_superate(14_999.99) == []
|
|
assert [s for s, _ in J.soglie_superate(15_000.0)] == [15_000.0]
|
|
|
|
|
|
def test_le_soglie_si_accumulano_e_restano_ordinate():
|
|
"""A $20k si devono presentare ENTRAMBE: la seconda non cancella la prima."""
|
|
assert [s for s, _ in J.soglie_superate(20_000.0)] == [15_000.0, 20_000.0]
|
|
|
|
|
|
def test_equity_non_leggibile_non_e_una_soglia_superata():
|
|
"""P5: «non vedo» non e' «va tutto bene» — e non e' nemmeno «la soglia e' arrivata»."""
|
|
assert J.soglie_superate(None) == []
|
|
|
|
|
|
def test_ogni_soglia_dice_COSA_decidere_non_solo_che_e_arrivata():
|
|
"""P4: un'allerta risponde a due domande. Una riga che dice solo «sei a $15k» non serve."""
|
|
for _, testo in J.SOGLIE_CAPITALE:
|
|
assert len(testo) > 80 and "**" in testo, "la soglia non dice quale decisione sblocca"
|