PythagorasGoal/tests/test_journal.py

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