"""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))