journal: dichiarati nello stesso intervallo si sommano, giro corrente, finestra 7gg, DD netto (#27, #35)

#27: due movimenti dichiarati nello stesso intervallo si sommavano al posto di sovrascriversi
(l'aggancio "dichiarato+rilevato" ora cerca solo fonte=="rilevato").
#35 C9: salute() non conta piu' il giro dell'ora corrente prima che scatti (minuto :47 +
margine, MINUTO_GIRO costante di modulo).
#35 C15: la finestra "ultimi 7 giorni" della concentrazione ora copre 7 giorni di calendario,
non 8 (baseline = ultima lettura al o prima di giorno-7 fine giornata).
#35 C7: il drawdown si calcola sulla serie equity al netto dei movimenti di capitale certi,
cosi' un prelievo non si legge come perdita di trading e un versamento non alza il picco.

Test nuovi in tests/test_journal_u4.py; tests/test_journal.py aggiornato (l'orologio del
test su salute() era reale e non deterministico, ora e' fissato).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-27 13:35:08 +00:00
parent 1a7f670966
commit ac21e1c97d
3 changed files with 264 additions and 17 deletions
+49 -13
View File
@@ -248,12 +248,26 @@ def movimenti_capitale(con, fino_ts: str, rows=None) -> dict:
+ (f" il {d['dichiarato_il']}" if d.get("dichiarato_il") else "")
+ (f": {d['nota']}" if d["nota"] else "")
+ (f" (banda ±${d['banda']:,.2f})" if d["banda"] else ""))
gia = next((e for e in eventi if e["ts_dopo"] == cur["ts_utc"]), None)
if gia is not None: # rilevato E dichiarato: l'importo e' quello dichiarato
gia.update(delta=d["delta"], pct=d["delta"] / float(prev["equity"]) if prev["equity"] else None,
# (#27) l'aggancio "dichiarato+rilevato" vale SOLO contro un evento rilevato: se si
# cercasse un qualunque evento sullo stesso ts_dopo, il secondo dichiarato nello stesso
# intervallo troverebbe il primo (gia' "dichiarato") e lo SOVRASCRIVEREBBE, perdendo
# il primo importo invece di somarli.
rilevato = next((e for e in eventi
if e["ts_dopo"] == cur["ts_utc"] and e["fonte"] == "rilevato"), None)
if rilevato is not None: # rilevato E dichiarato: l'importo e' quello dichiarato
rilevato.update(delta=d["delta"], pct=d["delta"] / float(prev["equity"]) if prev["equity"] else None,
classe="movimento", fonte="dichiarato+rilevato",
dopo=float(prev["equity"]) + d["delta"], dichiarazione=nota_d)
continue
gia = next((e for e in eventi if e["ts_dopo"] == cur["ts_utc"]
and e["fonte"] in ("dichiarato", "dichiarato+rilevato")), None)
if gia is not None: # un altro dichiarato nello stesso intervallo: si SOMMA
nuovo_delta = gia["delta"] + d["delta"]
gia.update(delta=nuovo_delta,
pct=(nuovo_delta / float(gia["prima"]) if gia["prima"] else None),
dopo=float(gia["prima"]) + nuovo_delta,
dichiarazione=(gia["dichiarazione"] + "; " + nota_d))
continue
eventi.append(dict(ts_prima=prev["ts_utc"], ts_dopo=cur["ts_utc"],
prima=float(prev["equity"]), dopo=float(prev["equity"]) + d["delta"],
delta=d["delta"],
@@ -376,16 +390,23 @@ def pnl_giorno(con, giorno: date) -> dict:
twr_da_arming=(r_twr["twr"] if cum is not None else None), twr_motivo=r_twr["motivo"])
MINUTO_GIRO = 47 # deve restare d'accordo con la crontab e scripts/cron_book.sh (#35 C9)
def salute(con, giorno: date) -> dict:
"""Giri attesi: 24 per un giorno CHIUSO, le ore trascorse per quello in corso.
Un giorno in corso confrontato con 24 produce un allarme a ogni esecuzione — e un
allarme che scatta sempre e' un allarme che non verra' letto il giorno che e' vero.
Il giro dell'ora CORRENTE scatta al minuto `MINUTO_GIRO` (il cron e' `47 * * * *`): prima
di quel minuto non e' ancora avvenuto, e contarlo comunque (#35 C9) segnalava un giro
"mancante" che semplicemente non era ancora il suo turno.
"""
lb = stato_libro(con, giorno)
adesso = datetime.now(timezone.utc)
in_corso = giorno == adesso.date()
atteso = (adesso.hour + 1) if in_corso else 24
atteso = (adesso.hour + (1 if adesso.minute >= MINUTO_GIRO + 1 else 0)) if in_corso else 24
fatti = lb.get("giri", 0)
return dict(giri_book=fatti, feed_skh_min=lb.get("feed_skh_min"), atteso_giri=atteso,
giorno_in_corso=in_corso, giri_mancanti=max(0, atteso - fatti))
@@ -591,16 +612,19 @@ def lettura(con, voce: dict) -> list[dict]:
# il cumulato di TRADING: i movimenti di capitale certi escono, o un versamento
# diventa "concentrazione di P&L" (e' successo: 25/08, +$1.399 letti come profitto)
mv_ev = [e for e in (p.get("movimenti_arming") or []) if e["classe"] == "movimento"]
g7s = (date.fromisoformat(voce["giorno"]) - timedelta(days=7)).isoformat()
tot = e1 - e0 - sum(e["delta"] for e in mv_ev)
picco = max(r["equity"] for r in eq)
g7 = [r for r in eq if r["ts_utc"][:10] >= g7s]
mv7 = sum(e["delta"] for e in mv_ev if e["ts_dopo"][:10] >= g7s)
# (#35 C15) "ultimi 7 giorni" deve essere VERO: la baseline e' l'ULTIMA lettura al o
# prima del giorno-7 a fine giornata — non la PRIMA lettura di quel giorno. Filtrare per
# data (>= giorno-7) include anche le ore del giorno-7 gia' trascorse, cioe' 8 giorni di
# calendario, non 7.
g7_cutoff = (date.fromisoformat(voce["giorno"]) - timedelta(days=7)).isoformat() + "T23:59:59+00:00"
base7 = next((r for r in reversed(eq) if r["ts_utc"] <= g7_cutoff), None)
mv7 = sum(e["delta"] for e in mv_ev if e["ts_dopo"] > g7_cutoff)
# Soglia di rilevanza: su un cumulato di pochi dollari la "concentrazione" e' aritmetica
# vera e informazione zero. Una regola che si accende su $2 insegna a saltare la sezione.
rilevante = abs(tot) >= max(5.0, 0.01 * e1)
if rilevante and g7:
quota = (e1 - g7[0]["equity"] - mv7) / tot
if rilevante and base7 is not None:
quota = (e1 - base7["equity"] - mv7) / tot
if quota > 0.60:
coda = ("Un risultato concentrato in una finestra non e' un tasso di rendimento: "
"e' un evento.")
@@ -609,10 +633,22 @@ def lettura(con, voce: dict) -> list[dict]:
"senza questi 7 giorni il libro sarebbe sotto. " + coda)
dire("concentrazione", f"**Il {100*quota:.0f}% di tutto il P&L cumulato "
f"(${tot:+,.2f}) viene dagli ultimi 7 giorni.** " + coda, "nota")
if picco > 0:
dd = (e1 - picco) / picco
# (#35 C7) il DD si calcola sulla serie AL NETTO dei movimenti CERTI (gli stessi che il
# TWR scorpora): sulla serie grezza un prelievo si legge come drawdown di trading e un
# versamento alza il picco, mascherando un DD vero appena prima del versamento.
mv_ord = sorted(mv_ev, key=lambda e: e["ts_dopo"])
cum_mov, i_mov, eq_netto = 0.0, 0, []
for r in eq:
while i_mov < len(mv_ord) and r["ts_utc"] >= mv_ord[i_mov]["ts_dopo"]:
cum_mov += mv_ord[i_mov]["delta"]
i_mov += 1
eq_netto.append(r["equity"] - cum_mov)
picco_netto = max(eq_netto)
if picco_netto > 0:
dd = (eq_netto[-1] - picco_netto) / picco_netto
if dd < -0.005:
dire("drawdown", f"Equity **{100*dd:.2f}%** sotto il picco (${picco:,.2f}). "
dire("drawdown", f"Equity **{100*dd:.2f}%** sotto il picco (${picco_netto:,.2f}, "
"al netto dei movimenti di capitale certi). "
"⚠️ e' un DD su letture ORARIE, non sul minimo intra-giorno.", "nota")
# --- volatilita' implicita contro realizzata ---------------------------------------------
+8 -4
View File
@@ -31,12 +31,16 @@ def db(tmp_path):
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))
class _OraFissa(datetime): # ora ferma: il test non dipende da quando gira
@classmethod
def now(cls, tz=None):
return datetime(2026, 8, 25, 14, 50, tzinfo=timezone.utc)
monkeypatch.setattr(J, "datetime", _OraFissa)
oggi = _OraFissa.now(timezone.utc).date()
monkeypatch.setattr(J, "stato_libro", lambda c, g: dict(giri=15, 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["atteso_giri"] == 15 # ore 14:50, il giro delle 14 (:47) e' scattato
assert s["giri_mancanti"] == 0
+207
View File
@@ -0,0 +1,207 @@
"""Test nuovi per l'unita' U4 (issue #27, #35 C9/C15/C7) — journal.py.
Non ripete la suite esistente: la tocca solo dove il comportamento CAMBIA.
"""
from __future__ import annotations
import json
import sys
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
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()
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 _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 _dichiara(monkeypatch, tmp_path, righe):
f = tmp_path / "mov.jsonl"
f.write_text("\n".join(json.dumps(r) for r in righe) + "\n")
monkeypatch.setattr(J, "MOVIMENTI_DICHIARATI", f)
return f
# ============================================================================================
# #27 — due dichiarati nello stesso intervallo si SOMMANO, non si sovrascrivono
# ============================================================================================
def test_due_dichiarati_nello_stesso_intervallo_si_sommano(db, monkeypatch, tmp_path):
"""+25 e +30 dichiarati nello stesso intervallo, trading reale +5: prima del fix il
secondo dichiarato sovrascriveva il primo (delta finale 30, non 55)."""
_serie_equity(db, [("2026-08-25T07:07:01+00:00", 1000.0),
("2026-08-25T08:07:01+00:00", 1060.0)]) # +60 = 25+30 mov. +5 trading
_dichiara(monkeypatch, tmp_path, [
dict(ts_utc="2026-08-25T07:20:00+00:00", delta_usd=25.0, nota="primo versamento"),
dict(ts_utc="2026-08-25T07:40:00+00:00", delta_usd=30.0, nota="secondo versamento")])
r = J.rendimento_twr(db, "2026-08-25T23:59:59+00:00")
assert len(r["eventi"]) == 1 # UN evento, non due
e = r["eventi"][0]
assert e["fonte"] == "dichiarato" and e["delta"] == pytest.approx(55.0) # 25+30, non 30
assert e["dopo"] == pytest.approx(1055.0)
assert "primo versamento" in e["dichiarazione"] and "secondo versamento" in e["dichiarazione"]
assert r["certi"] == pytest.approx(55.0)
assert r["trading"] == pytest.approx(5.0) # 1060-1000-55
# solo due letture: l'intero intervallo e' il movimento, nessun segmento di trading — 0%
# e' il limite DICHIARATO di rendimento_twr ("nessun segmento = nessun tempo a mercato"),
# non un valore inventato.
assert r["twr"] == pytest.approx(0.0)
def test_dichiarato_su_rilevato_resta_come_prima_anche_con_un_altro_dichiarato_altrove(db, monkeypatch, tmp_path):
"""Non regressione: l'aggancio dichiarato+rilevato non deve rompersi perche' ora si guarda
anche ad altri dichiarati."""
monkeypatch.setattr(J, "_px_1h", lambda a: __import__("pandas").Series(
100.0, index=__import__("pandas").date_range("2026-08-20", "2026-08-27", freq="1h", tz="UTC")))
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 667.49),
("2026-08-25T11:47:01+00:00", 2066.88),
("2026-08-25T12:47:01+00:00", 2064.56)])
_dichiara(monkeypatch, tmp_path, [dict(ts_utc="2026-08-25T11:20:00+00:00", delta_usd=1399.39,
nota="bonifico")])
r = J.rendimento_twr(db, "2026-08-25T23:59:59+00:00")
assert len(r["eventi"]) == 1
e = r["eventi"][0]
assert e["fonte"] == "dichiarato+rilevato" and e["delta"] == pytest.approx(1399.39)
# ============================================================================================
# #35 C9 — il giro dell'ora corrente conta solo dopo che e' scattato (minuto :47 + margine)
# ============================================================================================
def test_atteso_giri_al_minuto_10_non_conta_ancora_il_giro_dell_ora_corrente(db, monkeypatch):
class _OraFissa(datetime):
@classmethod
def now(cls, tz=None):
return datetime(2026, 8, 25, 14, 10, tzinfo=timezone.utc)
monkeypatch.setattr(J, "datetime", _OraFissa)
monkeypatch.setattr(J, "stato_libro", lambda c, g: dict(giri=14, feed_skh_min=0))
s = J.salute(db, date(2026, 8, 25))
assert s["atteso_giri"] == 14 # il giro delle 14 (:47) non e' ancora scattato
assert s["giri_mancanti"] == 0
def test_atteso_giri_al_minuto_50_conta_il_giro_dell_ora_corrente(db, monkeypatch):
class _OraFissa(datetime):
@classmethod
def now(cls, tz=None):
return datetime(2026, 8, 25, 14, 50, tzinfo=timezone.utc)
monkeypatch.setattr(J, "datetime", _OraFissa)
monkeypatch.setattr(J, "stato_libro", lambda c, g: dict(giri=15, feed_skh_min=0))
s = J.salute(db, date(2026, 8, 25))
assert s["atteso_giri"] == 15 # dopo le :47 il giro delle 14 e' scattato
assert s["giri_mancanti"] == 0
# ============================================================================================
# #35 C15 — "ultimi 7 giorni" deve coprire 7 giorni di calendario, non 8
# ============================================================================================
def test_la_finestra_di_concentrazione_e_di_7_giorni_non_8(db):
base = date(2026, 6, 24) # base .. base+13 = 14 giorni prima del giorno
righe = []
for i in range(7): # giorni 0..6: piatto
d = (base + timedelta(days=i)).isoformat()
righe += [(f"{d}T06:00:00+00:00", 1000.0), (f"{d}T18:00:00+00:00", 1000.0)]
d7 = (base + timedelta(days=7)).isoformat() # giorno-7: il salto avviene QUI, a meta' giornata
righe += [(f"{d7}T06:00:00+00:00", 1000.0), (f"{d7}T18:00:00+00:00", 1020.0)]
for i in range(8, 14): # giorni 8..13: piatto al nuovo livello
d = (base + timedelta(days=i)).isoformat()
righe += [(f"{d}T06:00:00+00:00", 1020.0), (f"{d}T18:00:00+00:00", 1020.0)]
d14 = (base + timedelta(days=14)).isoformat() # il giorno della voce
righe += [(f"{d14}T06:00:00+00:00", 1020.0), (f"{d14}T18:00:00+00:00", 1100.0)]
_serie_equity(db, righe)
assert len(righe) == 30 # >24: la sezione si attiva
v = _voce(libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl(movimenti_arming=[]), giorno=d14)
out = J.lettura(db, v)
riga = next(r for r in out if r["id"] == "concentrazione")
# baseline corretta = ULTIMA lettura di giorno-7 (1020, alle 18:00), non la PRIMA (1000,
# alle 06:00): (1100-1020)/(1100-1000) = 80%. Con la finestra sbagliata (8 giorni di
# calendario) sarebbe (1100-1000)/100 = 100%.
assert "80%" in riga["testo"]
assert "100%" not in riga["testo"]
# ============================================================================================
# #35 C7 — il drawdown si legge sulla serie AL NETTO dei movimenti certi
# ============================================================================================
def test_un_prelievo_non_diventa_un_drawdown_di_trading(db):
base = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc)
punti = [base + timedelta(hours=i) for i in range(26)] # 26 letture, >24
for t in punti[:-1]:
db.execute("INSERT INTO equity VALUES (?,?,?)", (t.isoformat(), 1000.0, "t"))
db.execute("INSERT INTO equity VALUES (?,?,?)", (punti[-1].isoformat(), 800.0, "t"))
db.commit()
mov = dict(classe="movimento", delta=-200.0, ts_prima=punti[-2].isoformat(),
ts_dopo=punti[-1].isoformat(), prima=1000.0, dopo=800.0, pct=-0.2,
fonte="dichiarato", dichiarazione="prelievo di prova")
v = _voce(libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl(movimenti_arming=[mov]),
giorno=punti[-1].date().isoformat())
out = J.lettura(db, v)
# sulla serie grezza (picco 1000, ultima 800) sarebbe -20%: un prelievo, non una perdita.
assert "drawdown" not in _ids(out)
def test_un_versamento_non_nasconde_un_drawdown_di_trading_vero(db):
base = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc)
letture = [(base + timedelta(hours=i), 1000.0) for i in range(20)] # piatto, 20 letture
letture.append((base + timedelta(hours=20), 1500.0)) # versamento +500
for h, v in ((21, 1490.0), (22, 1480.0), (23, 1475.0), (24, 1470.0)): # trading in calo
letture.append((base + timedelta(hours=h), v))
for t, val in letture:
db.execute("INSERT INTO equity VALUES (?,?,?)", (t.isoformat(), val, "t"))
db.commit()
assert len(letture) == 25 # >24
mov = dict(classe="movimento", delta=500.0, ts_prima=letture[19][0].isoformat(),
ts_dopo=letture[20][0].isoformat(), prima=1000.0, dopo=1500.0, pct=0.5,
fonte="dichiarato", dichiarazione="versamento di prova")
v = _voce(libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl(movimenti_arming=[mov]),
giorno=letture[-1][0].date().isoformat())
out = J.lettura(db, v)
riga = next(r for r in out if r["id"] == "drawdown")
# netto: picco 1000 (prima/al versamento), ultima 970 (1470-500) -> -3.00%. Sulla serie
# grezza (picco 1500 gonfiato dal versamento) sarebbe -2.00%: sottostima il vero calo.
assert "-3.00%" in riga["testo"]
assert "-2.00%" not in riga["testo"]