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