Files
PythagorasGoal/tests/test_journal.py
T
Adriano Dal Pastro 2751a7efd0 journal: la voce del giorno IN CORSO non si presenta piu' come una giornata
cron_daily (00:30 UTC) faceva DUE chiamate al giornale: chiudeva IERI (giusto)
e apriva OGGI. La pagina di oggi nasceva con ~30 minuti dentro e nessuno la
aggiornava fino alla notte dopo.

Il difetto non era il dato mancante: era che la pagina non lo diceva dove si
legge. Aveva TUTTE le sezioni di una chiusa -> passava ogni controllo di
completezza e freschezza, e la parzialita' stava solo in §Salute, quinta
sezione su otto. Stessa famiglia di "una barra presente non e' una giornata
presente", in veste nuova: la pagina e' presente, il giorno no.

Taglia misurata sul caso reale di oggi: la pagina congelata alle 00:37 diceva
"giorno: $+0.54 (1 letture)" e la regola [pnl] chiosava "senza operare". Alle
08:54 il libro aveva fatto 3 fill, 2 round-trip e +$25.86 ($642.56 -> $667.88).
Avrebbe raccontato una giornata ferma per tutta una giornata operativa.

Riparato in due pezzi indipendenti:
  1. la pagina dichiara la parzialita' nel TITOLO e nella prima riga —
     "PARZIALE (giorno in corso)", copertura esplicita (N giri su 24), il fatto
     che non si aggiorna da sola, e quali numeri sono di quella frazione;
  2. tolta la seconda chiamata dal cron. In docs/journal/ restano solo giorni
     chiusi; lo stato corrente si prende da journal.py a mano (pagina marcata)
     o da trades.db, che cron_book sincronizza ogni ora.

Scelta dichiarata: NON si rigenera ogni ora. Costerebbe una riga in cron_book
ma riscriverebbe un file tracciato da git 24 volte al giorno per un consumatore
che non esiste (l'analista legge il giorno chiuso). Se un domani servisse, la
strada e' RIGENERARLA, non congelarla: e' scritto nel commento del cron.

Prove (test_journal 28 -> 31): una accende la marcatura, una la tiene spenta su
un giorno chiuso (una marcatura sempre accesa non si legge), una e' guardia sul
SORGENTE di cron_daily.sh — ogni chiamata a journal.py deve avere --giorno
esplicito, perche' senza argomenti scrive OGGI.

Regolare docs/journal/2026-08-25.md rigenerato: ora si dichiara PARZIALE (9/24).

NON riparato, e registrato come debito aperto in CLAUDE.md §5:
test_wave_0726::test_t1_canonical_riproduce_il_backtest_ufficiale fallisce, e
falliva gia' prima di questa modifica (verificato con git stash sull'albero
pulito). Sharpe hold-out SKH01 canonical 1,9223 contro la banda cablata
1.3 < h < 1.9: il codice non e' cambiato, sono cambiati i dati (data/raw e'
gitignored, il cron lo ricostruisce ogni notte). La banda NON e' stata
allargata — lezione 07/08. Suite: 710 passati, 1 fallito.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-25 08:59:42 +00:00

361 lines
17 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}"